/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "xdrbase.h"
#include "limit.h"
#include "jm.h"
#include "resreq.h"
#if defined(QUICKPOOL_WITH_CUDA)
#include "nvml.h"
#endif

#define PRE_EXECUTION_ABORT 99
extern char **environ;

#if defined(QUICKPOOL_WITH_CUDA)
extern bool_t isnvml_initialized(void);
#endif

static void execute_job(struct jobEntry *);
static void execute_restart_job(struct jobEntry *);
static int set_env_4job(struct jobEntry *);
static int send_results(struct jobEntry *);

static void copy_jobinfo(struct jobInfo *, struct jobInfo *);

static void execute_user_postcmd(struct jobEntry *);
static int auto_requeue_job(struct jobEntry *);
static int stop_job(struct jobEntry *, int, int);

static int setuid_4prepost_cmd(struct jobEntry *);

static int setup_cpu_bind(struct jobEntry *);
static int setup_gpu_option(struct jobEntry *);

/**
 * @brief       fork a child to execute the user job.
 *
 * @param[in]   job          #1: job entry to be executed
 *
 * @retval      0            #1: succeed
 * @retval      != 0         #2: fork fail
 *
 * @note        create cgroup job entry for the job.
 *              For child, we need close log and all channel fds.
 */
int
fork_2start_job(struct jobEntry *job)
{
    struct jobInfo *jobBase = &(job->job_base);

    char strJobId[32];
    log_message(DEBUG2, EXECUTE, "%s: start to run job[%s].",
                __func__, jobid_2string(jobBase->jobid, strJobId));

    jobBase->reasons = 0;
    jobBase->subreasons = 0;

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, "%s: fork() failed starting job %s: %m",
                    __func__, jobid_2string(jobBase->jobid, strJobId));

        return ERROR_FORK_FAIL;
    }

    if (0 == pid) { // child to execute job

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
                 baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
        open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                 baseParams[BASE_JM_TIME_DEBUG].int_value);

        if (setup_cpu_bind(job)) {
            log_message(ERR, NOCLASS, "%s: failed to bind numa core for job[%s], pid[%d]",
                        __func__, jobid_2string(jobBase->jobid, strJobId), pid);
            exit(-1);
        }

        execute_job(job); // child exit in this function
    }

    // parent, set job pid and create job cgroup
    jobBase->job_pid = pid;

    if (jobBase->submit->options & SUBOPT_RESTART) {
        jobBase->job_pgid = 0;
    } else {
        jobBase->job_pgid = pid;        
    }

    job->lost = FALSE;
    job->comm_failcnt = 0;
    job->update_usage = FALSE;

    //set cgroup entry, implement later
    int rc;
    rc = cgroup_setup_job_entry(jobBase->jobid, jobBase->job_pid, "memory");
    if (rc < 0) {
        log_message(ERR, NOCLASS, "%s: failed to set cgroup(memory) for job[%s], pid[%d]",
                    __func__, jobid_2string(jobBase->jobid, strJobId), pid);
    } else if (baseParams[CGROUP_MEMORY_ENFORCE].bool_value) {
        /* enforce memory limit in cgroup when CGROUP_MEMORY_ENFORCE enable */
        setup_mem_cgroup(job);
    }
    rc = cgroup_setup_job_entry(jobBase->jobid, jobBase->job_pid, "cpuacct");
    if (rc < 0) {
        log_message(ERR, NOCLASS, "%s: failed to set cgroup(cpuacct) for job[%s], pid[%d]",
                    __func__, jobid_2string(jobBase->jobid, strJobId), pid);
    }

    return 0;
} // end function fork_2start_job

/**
 * @brief       Tell JS that job is finished.
 *
 * @param[in]   job          #1: job entry
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function is called twice for a job.
 *              1. Job itself is finished, tell JS that job is DONE or EXIT.
 *              2. Job post scrtip is finished, tell JS that post script is DONE or EXIT.
 *              If job control script is working, do nothing, if job post
 *              script has been started, delete the job entry.
 */
int
job_finish(struct jobEntry *job)
{
    if (job->job_base.act_pid) {
        return 0;
    }

    if (1 < job->job_base.number_node) {
        read_job_acct(job);
    }

    if (status_job(BATCH_STATUS_JOB, job, job->job_base.status,
                   (job->job_base.start_time > gtimeNodeBoot) ? 0 : ERROR_HOST_REBOOT) < 0) {
        job->comm_failcnt ++;
        return -1;
    }

    if ((job->job_base.status & JOB_STATUS_PEND)
        && (REASON_JM_ARGUMENT == job->job_base.reasons
            || REASON_JM_CREATE_JOBFILE == job->job_base.reasons)) {

        delete_jmjob(job);
        return 0;
    }

    if (job->post_started) {

        delete_jmjob(job);
        return 0;
    }

    char strJobId[32];

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(job->job_base.jobid, strJobId), "fork");
        job->comm_failcnt ++;
        return -1;
    }

    if (0 == pid) {

        bool_t doSendResults;
        doSendResults = (!job->status_reset && !(job->job_base.status & JOB_STATUS_PEND));

        gf_setenv("QP_EXECUTE", (char *)"END");

        if (prepare_4postcmd(job) == -1 && doSendResults) {

            change_uid(guidManager);
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
	                jobid_2string(job->job_base.jobid, strJobId), "prepare_4postcmd");
            exit(-1);
        }

        int hasError = 0;
        if (doSendResults) {

            if (send_results(job) != 0) {
                hasError = -1;
                log_message(ERR, NOCLASS, "%s: has error send result for job<%s>",
                            __func__, jobid_2string(job->job_base.jobid, strJobId));
            }
        }

        if (NULL != job->job_base.submit->postjob_command
            && '\0' != job->job_base.submit->postjob_command[0]) {

            execute_user_postcmd(job);
        }        

        if (execute_queue_postcmd(job) != 0) {
            hasError = -1;
            log_message(ERR, NOCLASS, "%s: failed to run queue post for job %s", __func__,
                        jobid_2string(job->job_base.jobid, strJobId));
        }

        if (job->job_base.reasons & EXIT_REQUEUE) {
            job->job_base.reasons &= ~EXIT_REQUEUE;
        }

        if ((!job->status_reset) && (!((job->job_base.reasons == REASON_QUEUE_PRESCRIPT)
                                     && (job->job_base.status & JOB_STATUS_PEND)))) {

            if (remove_jobfile(job) != 0) {
                hasError = -1;
                log_message(ERR, NOCLASS, "%s: has error to remove buffer file for job<%s>",
                            __func__, jobid_2string(job->job_base.jobid, strJobId));
            }
        }

        exit(hasError);
    }

    if (job->succeed) {
        job->post_started = TRUE;
        job->job_base.job_pid = pid;
        job->job_base.job_pgid = pid;
        job->job_base.status = JOB_STATUS_RUN;
    } else {
        delete_jmjob(job);
    }

    return 0;
} // end function job_finish

/**
 * @brief       Put the job into JM job list.
 *
 * @param[in]   job          #1: job entry
 *
 * @note        Job is sorted by specified priority and started time.
 */
void
putjob_injmJL(struct jobEntry *job)
{
    struct jobEntry *loopJob;

    for (loopJob = (struct jobEntry *)glistJmJobs->forw; loopJob != (struct jobEntry *)glistJmJobs;
         loopJob = (struct jobEntry *)loopJob->forw) {

        if (loopJob == job) {
            return;
        }

        if (job->job_base.submit->spec_priority < loopJob->job_base.submit->spec_priority) {
            break;
        }

        if ((job->job_base.submit->spec_priority == loopJob->job_base.submit->spec_priority)
             && (job->job_base.start_time - loopJob->job_base.start_time) >= 0) {
            break;
        }
    }

    gf_list_insert(glistJmJobs, (LIST_T *)loopJob, (LIST_T *)job);

    return;
} // end function putjob_injmJL

/**
 * @brief       create JM job.
 *
 * @param[in|out]  job          #1: job entry
 * @param[in]      jobbase      #2: job info from JS
 * @param[out]     reply        #3: error no
 *
 * @retval         0            #1: succeed
 * @retval         -1           #2: failed
 *
 * @note        Create JM job and put it in JM job list.
 */
int
create_jmjob(struct jobEntry *job, struct jobInfo *jobbase, int *reply)
{
    *reply = 0;

    if (NULL != jobbase->run_window && '\0' != jobbase->run_window[0]) {
        parse_time_window(jobbase->run_window, job->job_base.run_winlist, "queue");
    }

    char strJobId[32];
    if (jobbase->resume_condition && jobbase->resume_condition[0] != '\0') {

        job->resume_cond = check_threshold(jobbase->resume_condition);
        if (NULL == job->resume_cond) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                        jobid_2string(job->job_base.jobid, strJobId),
                        "check_threshold", "resume_condition");
            *reply = ERROR_ARGUMENT;
            return -1;
        }
    }

    if (jobbase->stop_condition && jobbase->stop_condition[0] != '\0') {

        job->stop_cond = check_threshold(jobbase->stop_condition);
        if (NULL == job->stop_cond) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                        jobid_2string(job->job_base.jobid, strJobId),
                        "check_threshold", "stop_condition");
            *reply = ERROR_ARGUMENT;
            return -1;
        }
    }

    time_t currentTime = time(NULL);

    if (0 != job->job_base.resource_limits[RESOURCE_LIMIT_RUN].rlim_curh) {
        job->job_base.resource_limits[RESOURCE_LIMIT_RUN].rlim_curl = 0x7fffffff;
    }
    job->running = 0;

    job->winclose_time = currentTime;
    job->mig_times = 1;

    job->inst_usage = (struct jobResourceUsage *)calloc(job->job_base.number_node,
                                                        sizeof(struct jobResourceUsage));
    if (NULL == job->inst_usage) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", job->job_base.number_node*sizeof(struct jobResourceUsage));
        relaunch();
    }

    job->max_usage = (struct jobResourceUsage *)calloc(job->job_base.number_node,
                                                       sizeof(struct jobResourceUsage));
    if (NULL == job->max_usage) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", job->job_base.number_node*sizeof(struct jobResourceUsage));
        relaunch();
    }

    unsigned int i;
    for (i=0; i<job->job_base.number_node; i++) {
        init_job_resourceusage(&(job->inst_usage[i]));
        init_job_resourceusage(&(job->max_usage[i]));
    }

    job->chkpnt_time = currentTime;
    job->actcmd_state = ACT_NO;

    log_message(DEBUG, EXECUTE, "%s: job[%s] options2=%x ",
                __func__, jobid_2string(job->job_base.jobid, strJobId),
                job->job_base.submit->options2);

    copy_jobinfo(&job->job_base, jobbase);

    sigset_t newmask, oldmask;
    sigemptyset(&newmask);
    sigaddset(&newmask, SIGCHLD);
    sigprocmask(SIG_BLOCK, &newmask, &oldmask);
    putjob_injmJL(job);
    sigprocmask(SIG_SETMASK, &oldmask, NULL);

    return 0;
} // end function create_jmjob

/**
 * @brief       delete JM job.
 *
 * @param[in|out]  job          #1: job entry
 *
 * @note        Remove JM job from JM job list and free the job.
 *              Also unlink the job tmp file.
 */
void
delete_jmjob(struct jobEntry *job)
{
    char tmpDirName[PATH_MAX+1];
    get_tmpdir_4job(tmpDirName, job->job_base.jobid, job->job_base.execute_uid);

    char strJobId[32];
    char fileBuf[PATH_MAX+1];
    snprintf(fileBuf, PATH_MAX, "%s/jobstatus.%s",
             tmpDirName, jobid_2longstring(job->job_base.jobid, strJobId));

    if (unlink(fileBuf) < 0 && errno != ENOENT) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                    jobid_2string(job->job_base.jobid, strJobId), "unlink", fileBuf);
    }

    snprintf(fileBuf, PATH_MAX, "%s/.%s.acct", tmpDirName, job->job_base.submit->script);
    if (unlink(fileBuf) < 0 && errno != ENOENT) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                    jobid_2string(job->job_base.jobid, strJobId), "unlink", fileBuf);
    }

    snprintf(fileBuf, PATH_MAX, "%s/%lld.ee.acct", tmpDirName, job->job_base.jobid);
    if (unlink(fileBuf) < 0 && errno != ENOENT) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                    jobid_2string(job->job_base.jobid, strJobId), "unlink", fileBuf);
    }

    /* Clean up cgroup enry for job. */
    cgroup_rm_job_entry(job->job_base.jobid, "memory");
    cgroup_rm_job_entry(job->job_base.jobid, "cpuacct");

    sigset_t newmask, oldmask;
    sigemptyset(&newmask);
    sigaddset(&newmask, SIGCHLD);
    sigprocmask(SIG_BLOCK, &newmask, &oldmask);
    gf_list_remove(glistJmJobs, (LIST_T *)job);
    sigprocmask(SIG_SETMASK, &oldmask, NULL);

    unsigned int i;
    for (i=0; i<job->job_base.number_node; i++) {

        snprintf(fileBuf, PATH_MAX, "%s/.%s.%d.acct",
                 tmpDirName, job->job_base.submit->script, i);
        if (unlink(fileBuf) < 0 && errno != ENOENT) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                        jobid_2string(job->job_base.jobid, strJobId), "unlink", fileBuf);
        }

        if (0 < job->inst_usage[i].npgids) {
            FREEUP(job->inst_usage[i].pgid);
        }

        if (0 < job->inst_usage[i].npids) {
            FREEUP(job->inst_usage[i].pidinfo);
        }

        if (0 < job->max_usage[i].npgids) {
            FREEUP(job->max_usage[i].pgid);
        }

        if (0 < job->max_usage[i].npids) {
            FREEUP(job->max_usage[i].pidinfo);
        }
    }

    if (rmdir(tmpDirName) < 0 && errno != ENOENT) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                    jobid_2string(job->job_base.jobid, strJobId), "rmdir", tmpDirName);
    }

    gf_xdr_free(libxdr_jobbase, &job->job_base);
    FREEUP(job->inst_usage);
    FREEUP(job->max_usage);

    free_requirement(job->resume_cond);
    free_requirement(job->stop_cond);
    FREEUP(job->resume_cond);
    FREEUP(job->stop_cond);
    FREEUP(job);

    return;
} // end function delete_jmjob

struct requirement *
check_threshold(char *strresreq)
{
    struct requirement *req;
    req = (struct requirement *)calloc(1, sizeof(struct requirement));
    if (NULL == req) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct requirement));
        relaunch();
    }

    if (set_requirement(strresreq, gptrSharedConf->qpinfo, REQ_SELECT, req) != 0) {

        free_requirement(req);
        FREEUP(req);
        log_message(DEBUG, EXECUTE, "%s: failed to parse threshold requirement(%s).",
                    __func__, strresreq);

        return NULL;
    }

    return req;
} // end function check_threshold

/**
 * @brief       Update job status or usage to JS.
 *
 * @param[in]   session      #1: session time
 *
 * @note        If send status failed before, we would try send it.
 *              Otherwise, update job usage to JS.
 */
void
status_report(time_t session)
{
    struct jobEntry *job, *next;

    for (job = (struct jobEntry *)glistJmJobs->back; job != (struct jobEntry *)glistJmJobs;
         job = next) {
        struct jobInfo *jobBase = &job->job_base;

        next = (struct jobEntry *)job->back;

        char strJobId[32];
        log_message(DEBUG, EXECUTE, "%s: checking job %s comm_failcnt %d lost %s update_usage %s session time %ld start_time %ld status %x",
                    __func__, jobid_2string(jobBase->jobid, strJobId), job->comm_failcnt,
                    job->lost?"true":"false", job->update_usage?"true":"false", session,
                    jobBase->start_time, jobBase->status);

        if (0 > job->comm_failcnt) {
            continue;
        }

        if (job->lost) {
            continue;
	}

        if (0 == job->comm_failcnt && !job->update_usage) {
            continue;
	}

        if ((session >= jobBase->start_time) && (session < jobBase->start_time + 10)) {

            /* don't retry other jobs either */
            if (!IS_JOB_STARTED(jobBase->status)) {
                continue;
            }
        }

        js_request_t reqCode;
        if (0 == job->comm_failcnt && job->update_usage) {
            reqCode = BATCH_RUSAGE_JOB;
        } else {
            reqCode = BATCH_STATUS_JOB;
        }

        int ret;
        ret = status_job(reqCode, job, jobBase->status, 0);
        if (BATCH_RUSAGE_JOB == reqCode) {
            continue;
        }

        if (0 <= ret) {
            if (0 < job->comm_failcnt) {
                job->comm_failcnt = 0;
            }
        } else {
            job->comm_failcnt ++;
        }
    }

    return;
} // end function status_report

/**
 * @brief       Reset job status after JM/JS restart.
 *
 * @param[in]   job          #1: job entry
 *
 * @note        read job status from log and check if it is still alive.
 */
void
reset_jobstatus(struct jobEntry *job)
{
    struct jobInfo *jobBase = &job->job_base;
    int jmLogFound = jmread_jobstatus(job);

    int ret;
    if (jobBase->act_pid) {

        kill(-jobBase->act_pid, SIGCONT);
        ret = send_signal_2job(job, SIGCONT, FALSE);

        if (0 > ret) {
            jobpid_lost(job);
        }

        return;
    }

    int jsJobStatus = jobBase->status;
    int jsReasons = jobBase->reasons;
    int jsSubReasons = jobBase->subreasons;
    if (jsJobStatus & JOB_STATUS_EXIT) {

        if ((jsReasons == EXIT_ZOMBIE) && (jobBase->submit->options & SUBOPT_RERUNNABLE)) {

            job->status_reset = TRUE;
        }

        if (jobBase->job_pid == 0) {

            jobBase->status = jsJobStatus;
            return;
        }

        signal_job_begin(job, QUICKPOOL_SIGKILL_OTHER, 0, TRUE);

        return;
    }

    ret = 0;
    if (jmLogFound < 0) {

        switch (MASK_STATUS(jobBase->status)) {
        case JOB_STATUS_PEND:

            if (!(jsJobStatus & JOB_STATUS_PEND)) {
                job->comm_failcnt ++;
            }

            break;
        case JOB_STATUS_RUN:

            if (jobBase->job_pid == 0) {
                JM_SET_STATE(job, JOB_STATUS_PEND);
            } else {
                job->mig_times = 1;
            }

            if (!(jsJobStatus & JOB_STATUS_RUN)) {
                job->comm_failcnt ++;
            }

            break;
        case JOB_STATUS_SSUSP:
        case JOB_STATUS_USUSP:

            ret = stop_job(job, jobBase->reasons, jobBase->subreasons);

            if (!(IS_JOB_SUSPEND(jsJobStatus))) {
                job->comm_failcnt ++;
            }

            break;
        }
    } else {

        switch (MASK_STATUS(jsJobStatus)) {
        case JOB_STATUS_PEND:

            break;
        case JOB_STATUS_RUN:

            if (jobBase->job_pid == 0) {
                JM_SET_STATE(job, JOB_STATUS_PEND);
            } else {
                job->mig_times = 1;
                ret = take_resume_action(job, QUICKPOOL_SIGCONT_OTHER, TRUE);
            }

            break;
        case JOB_STATUS_SSUSP:
        case JOB_STATUS_USUSP:
            ret = stop_job(job, jsReasons, jsSubReasons);

            break;
        }
    }

    if (0 > ret) {
        jobpid_lost(job);
    }

    return;
} // end function reset_jobstatus

/**
 * @brief       Reset job nice priority.
 *
 * @param[in]   job          #1: job to reset its nice priority
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, setpriority failure
 *
 * @note        If job is modified, we need reset job priority.
 *              Or after jm restart, we need reset job priority. 
 */
int
renice_job(struct jobEntry *job)
{
    struct jobInfo *jobBase = &job->job_base;
    char strJobId[32];
    log_message(DEBUG, EXECUTE, "%s: Job <%s> nice value %d",
                __func__, jobid_2string(jobBase->jobid, strJobId), jobBase->nice);

    int which, who;
    if (0 == jobBase->number_node) {
        return 0;
    }

    if (0 == job->inst_usage[0].npgids) {
        if (jobBase->job_pgid == 0) {

            who = jobBase->job_pid;
            which = PRIO_PROCESS;
        } else {
            which = PRIO_PGRP;
            who = jobBase->job_pgid;
        }

        if (setpriority(which, who, jobBase->nice) == -1) {
            log_message(DEBUG, EXECUTE, "%s: prio_pgrp which %d who %d nice %d for job %s: %m",
	                __func__, which, who, jobBase->nice,
                        jobid_2string(jobBase->jobid, strJobId));

            which = PRIO_PROCESS;
            who = jobBase->job_pid;
            if (setpriority(which, who, jobBase->nice) == -1) {
                log_message(DEBUG, EXECUTE, "%s: prio_process which %d who %d nice %d for job %s: %m",
	                    __func__, which, who, jobBase->nice,
                            jobid_2string(jobBase->jobid, strJobId));
                return -1;
            }
        }

        return 0;
    }

    unsigned int i;
    which = PRIO_PGRP;
    for (i = 0; i < job->inst_usage[0].npgids; i++) {
        who = job->inst_usage[0].pgid[i];
        if (setpriority(which, who, jobBase->nice) == -1) {
            log_message(DEBUG, EXECUTE, "%s: prio_pgrp which %d who %d nice %d for job %s: %m",
                        __func__, which, who, jobBase->nice,
                        jobid_2string(jobBase->jobid, strJobId));
        }
    }

    return 0;
} // end function renice_job

/**
 * @brief       Job Pid has been gone.
 *
 * @param[in]   job          #1: job entry
 *
 * @note        When job pid is lost, set job to exit/done.
 */
void
jobpid_lost(struct jobEntry *job)
{
    struct jobInfo *jobBase = &job->job_base;

    char strJobId[32];
    log_message(DEBUG, EXECUTE, "%s: checking if job <%s> pid <%d> gone. lost %s w_status %d comm_failcnt %d post_started %d act_pid %d",
                __func__, jobid_2string(jobBase->jobid, strJobId), jobBase->job_pid,
                job->lost?"true":"false", job->w_status, job->comm_failcnt,
                job->post_started, jobBase->act_pid);

    if (!job->lost) {

        job->lost = TRUE;
        gboolCheckFinishJobs = TRUE;

        if (0 < job->comm_failcnt) {
            job->comm_failcnt = 0;
        }

        return;
    }

    if (!job->post_started) {

        char tmpDirName[PATH_MAX+1];
        get_tmpdir_4job(tmpDirName, job->job_base.jobid, job->job_base.execute_uid);

        change_uid(job->job_base.execute_uid);

        char strAcctFile[PATH_MAX+1];
        snprintf(strAcctFile, PATH_MAX, "%s/%lld.ee.acct", tmpDirName, job->job_base.jobid);

        FILE *fileAcct;
        fileAcct = fopen(strAcctFile, "r");
        if (NULL != fileAcct) {

            unsigned int lineNum = 0;
            struct record *taskRec;
            taskRec = read_file_line(fileAcct, &lineNum);
            if (NULL != taskRec) {

                int rank=0;
                unsigned int j, numTask;
                for (j= 0; j<taskRec->numKVs; j++) {
                    switch (taskRec->KVs[j].key) {
                    case RECORD_KEY_NUM_TASK:
                        numTask = atoi(taskRec->KVs[j].value);
                        if (numTask > job->job_base.number_node) {
                        }
                        break;
                    case RECORD_KEY_NODE_RANK:
                        rank = atoi(taskRec->KVs[j].value);
                        if (0 > rank) {
                            rank = 0;
                        }
                        break;
                    case RECORD_KEY_TASK_STATUS:
                        job->w_status = atoi(taskRec->KVs[j].value);
                        break;
                    case RECORD_KEY_USAGE_MEM:
                        job->inst_usage[rank].mem = atol(taskRec->KVs[j].value);
                        break;
                    case RECORD_KEY_USAGE_SWAP:
                        job->inst_usage[rank].swap = atol(taskRec->KVs[j].value);
                        break;
                    case RECORD_KEY_USAGE_UTIME:
                        job->inst_usage[rank].utime = atol(taskRec->KVs[j].value);
                        break;
                    case RECORD_KEY_USAGE_STIME:
                        job->inst_usage[rank].stime = atol(taskRec->KVs[j].value);
                        break;
                    }
                }

                job->execute_flag |= JOB_EXEC_FINISH;
            }

            fclose(fileAcct);
        }

        change_uid(guidManager);

        if (!(job->execute_flag & JOB_EXEC_FINISH)) {
            STATUS_WAIT_T wStatus;

            JOB_WSTATUS(wStatus) = 0xff00;
            job->w_status = JOB_WSTATUS(wStatus);
        }

        unsigned int i;
        for (i=0; i<jobBase->number_node; i++) {
            job->cpu_time = job->inst_usage[i].utime + job->inst_usage[i].stime;
        }
    }

    if (0 == jobBase->act_pid) {

        if (auto_requeue_job(job)) {

            JM_SET_STATE(job, JOB_STATUS_PEND);
            jobBase->reasons = REASON_AUTO_REQUEUE;
        } else {
            if (job->post_started) {
                if (job->w_status) {
                    JM_SET_STATE(job, JOB_STATUS_PERR);
                } else {
                    JM_SET_STATE(job, JOB_STATUS_PDONE);
                }
            } else {
                if (job->w_status) {
                    JM_SET_STATE(job, JOB_STATUS_EXIT);
                } else {
                    JM_SET_STATE(job, JOB_STATUS_DONE);
                }
            }
        }
    }

    job->comm_failcnt ++;

    return;
} // end function jobpid_lost

/**
 * @brief       Before start job, we need change user id to real execution user.
 *
 * @param[in]      jobid        #1: job id
 * @param[in]      user         #2: execution user
 * @param[in]      gid          #3: gid of the execution user
 * @param[in]      uid          #4: uid of the execution user
 *
 * @retval         0            #1: succeed
 * @retval         -1           #2: failed
 */
int
set_userid_4job(JOBID_T jobid, char *user, uid_t gid, uid_t uid)
{
    char strJobId[32];
    if (initgroups(user, gid) < 0) {
        log_message(ERR, NOCLASS, "%s: initgroups() jobid %s user %s failed: %m",
                    __func__, jobid_2string(jobid, strJobId), user);
        return -1;
    }

    if (setgid(gid) < 0) {
        log_message(ERR, NOCLASS, "%s: setgid() failed jobid %s gid %d",
                    __func__, jobid_2string(jobid, strJobId), gid);
        return -1;
    }

    if (setuid(uid) < 0) {
        log_message(ERR, NOCLASS, "%s: setuid() failed jobid %s user %s %m",
                    __func__, jobid_2string(jobid, strJobId), user);
        return -1;
    }

    return 0;
} // end function set_userid_4job

/**
 * @brief       Set job start argument.
 *
 * @param[in]   fromnode     #1: where job comes from 
 * @param[in]   execargv     #2: job argument
 *
 * @retval      not NULL     #1: arguments for starting the job
 * @retval      NULL         #2: failed, ee not found
 *
 * @note        We use ee to start the job, this function is to 
 *              wrap job starting argument into ee.
 */
char **
set_startjob_arguments(char *fromnode, char **execargv)
{
    int i = 0;

    char path[PATH_MAX+1];
    char *strPathEE;
    strPathEE = get_daemon_path("/ee", baseParams[BASE_SERVERDIR].string_value, path);
    if (NULL == strPathEE) {
        return NULL;
    }

    static char *argv[64];
    argv[i++] = path;
    argv[i++] = (char *)"-d";
    argv[i++] = (char *)baseParams[BASE_CONFDIR].string_value;

    if (NULL != fromnode) {
        argv[i++] = (char *)"-m";
        argv[i++] = fromnode;
    } else {
        argv[i++] = (char *)"-t";
    }

    for (; *execargv; execargv++) {
        argv[i++] = *execargv;
    }

    argv[i] = NULL;

    return argv;
} // end function set_startjob_arguments

/**
 * @brief       Do work after jm fork child and before it execute command.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
int
prepare_4postcmd(struct jobEntry *job)
{
    Signal_(SIGCHLD, SIG_DFL);
    Signal_(SIGTERM, SIG_IGN);
    Signal_(SIGINT,  SIG_IGN);

    close_log();
    gf_child_chan_close(-1);
    finalize_tcl();

    char logfile[PATH_MAX+1];
    snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
             baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
    open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
             baseParams[BASE_JM_TIME_DEBUG].int_value);

    struct jobInfo *jobBase = &job->job_base;
    char strJobId[32];
    if (setsid() == -1) {

        if (getpid() != getpgrp()) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
	                jobid_2string(jobBase->jobid, strJobId), "setsid");
        }
    }

    if (set_env_4job(job) < 0) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "set_env_4job");
        return -1;
    }

    if (jobBase->execute_user[0] == '\0') {

        if (set_execute_user(job) < 0) {
            change_uid(jobBase->execute_uid);
            log_message(DEBUG, EXECUTE, "%s: set_execute_user() failed for job <%s>",
	                __func__, jobid_2string(jobBase->jobid, strJobId));
            change_uid(guidManager);
            return -1;
        }

        gf_setenv("USER", jobBase->execute_user);
    }

    if (initgroups(jobBase->execute_user, job->execute_gid) < 0) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "initgroups", jobBase->execute_user);
    }

    if (setgid(job->execute_gid) < 0) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_D_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "setgid", job->execute_gid);
    }

    change_uid(jobBase->execute_uid);

    if (setup_path_4job(job, NULL) < 0) {
        if (!(jobBase->status & JOB_STATUS_PREJOB) && REASON_QUEUE_PRESCRIPT != jobBase->reasons) {

            change_uid(guidManager);
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                        jobid_2string(jobBase->jobid, strJobId), "setup_path_4job");
            change_uid(jobBase->execute_uid);
        }

        return -2;
    }

    return 0;
} // end function prepare_4postcmd

/**
 * @brief       To execute queue defined post script.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
int
execute_queue_postcmd(struct jobEntry *job)
{
    change_uid(guidManager);

    struct jobInfo *jobBase = &job->job_base;

    char strJobId[32];
    jobid_2string(jobBase->jobid, strJobId);

    log_message(DEBUG, EXECUTE, "%s: queue's post-command %s execute_flag %x for job %d status %x",
                __func__, jobBase->postqueue_command ? jobBase->postqueue_command : "NULL",
                job->execute_flag, strJobId, job->w_status);
    change_uid(jobBase->execute_uid);

    if (!jobBase->postqueue_command || jobBase->postqueue_command[0] == '\0') {
        return 0;
    }

    Signal_(SIGCHLD, SIG_DFL);

    sigset_t new_mask, old_mask;
    sigemptyset(&new_mask);
    sigaddset(&new_mask, SIGCHLD);
    sigprocmask(SIG_SETMASK, &new_mask, &old_mask);

    pid_t pid;
    pid = fork();
    if (0 > pid) {

        change_uid(guidManager);
        log_message(ERR, NOCLASS, "%s: fork() failed for job <%s> due to %m",
                    __func__, strJobId);
        change_uid(jobBase->execute_uid);
        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return -1;
    }

    if (0 < pid) {

        if (waitpid(pid, NULL, 0) < 0) {

	    change_uid(guidManager);
	    log_message(ERR, NOCLASS, "%s: waitpid(%d) failed for job <%s> due to %m",
	              __func__, pid, strJobId);
	    change_uid(jobBase->execute_uid);
            sigprocmask(SIG_SETMASK, &old_mask, NULL);

            return -1;
        }

        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return 0;
    }

    if (0 > chdir("/tmp")) {
        log_message(ERR, NOCLASS, "%s: failed to chkdir(/tmp) for job <%s> due to %m",
                    __func__, strJobId);
    }

    change_uid(guidManager);

    close_log();
    gf_child_chan_close(-1);
    finalize_tcl();

    char logfile[PATH_MAX+1];
    snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
             baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
    open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
             baseParams[BASE_JM_TIME_DEBUG].int_value);

    if (setuid_4prepost_cmd(job) < 0) {
        log_message(ERR, NOCLASS, "%s: queue's post-exec failed for job <%s>",
                    __func__, strJobId);
        exit(-1);
    }

    int i;
    for (i = 1; i < NSIG; i++) {
        Signal_(i, SIG_DFL);
    }

    sigset_t newmask;
    sigemptyset(&newmask);
    sigprocmask(SIG_SETMASK, &newmask, NULL);
    alarm(0);

    gf_setenv("PATH", (char *)"/bin:/usr/bin:/sbin:/usr/sbin");

    i = open("/dev/null", O_RDWR, 0);
    dup2(i, 0);
    dup2(i, 1);
    dup2(i, 2);

    char val[12];
    snprintf(val, 12, "%d", job->w_status);
    gf_setenv("QP_JOBEXIT_STAT", val);

    if (jobBase->status & JOB_STATUS_PEND)
        gf_setenv("QP_JOBPEND", (char *)" ");

    char *myargv[6];
    myargv[0] = (char *)"/bin/sh";
    myargv[1] = (char *)"-c";
    myargv[2] = jobBase->postqueue_command;
    myargv[3] = NULL;

    execvp("/bin/sh", myargv);

    log_message(ERR, NOCLASS, "%s: queue's post-exec command %s failed for job %s: %m.",
                __func__, jobBase->postqueue_command, strJobId);

    exit(-1);
} // end function execute_queue_postcmd

/**
 * @brief       Create job tmp dir.
 *
 * @param[out]  dirname      #1: tmp dir for the job
 * @param[in]   jobid        #2: job id
 * @param[in]   uid          #3: user who create the dir
 *
 * @note        Tmp files are used by the job under this directory.
 *              Following tmp files are created during job working life:
 *              jobstatus.jobid, .jobfile.acct, jobid.ee.acct
 */
void
get_tmpdir_4job(char *dirname, JOBID_T jobid, uid_t uid)
{
    memset(dirname, 0, PATH_MAX+1);
    
    strncpy(dirname, gstrTmpDir, PATH_MAX);
    strncat(dirname, "/", PATH_MAX);

    strncat(dirname, ".", PATH_MAX);
    strncat(dirname, gptrMyCluster->name, PATH_MAX);
    strncat(dirname, ".tmpdir", PATH_MAX);
    strncat(dirname, ".", PATH_MAX);

    char strJobId[32] = {0};
    snprintf(strJobId, 32, "%lld", jobid);
    strncat(dirname, strJobId, PATH_MAX);

    change_uid(uid);

    memset(strJobId, 0, sizeof(strJobId));
    if (access(dirname, W_OK) < 0) {
        mode_t previousUmask;

        previousUmask = umask(077);
        if ((mkdir(dirname, 0700) < 0) && errno != EEXIST) {

            change_uid(guidManager);

            log_message(ERR, NOCLASS, "%s: Unable to create job tmp directory %s for job %s",
                        __func__, dirname, jobid_2string(jobid, strJobId));
            dirname[0] = '\0';
        }
        umask(previousUmask);
    }

    change_uid(guidManager);

    return;
} // end function get_tmpdir_4job

/**
 * @brief       Record job status for jm recovering.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 *
 * @note        Job status changing is logged in tmp file jobstatus.jobid
 *              for jm recovery.
 */
int
jmlog_newstatus(struct jobEntry *job)
{
    int eno;
    struct jobInfo *jobBase = &(job->job_base);

    char tmpDirName[PATH_MAX+1];
    get_tmpdir_4job(tmpDirName, job->job_base.jobid, job->job_base.execute_uid);

    change_uid(jobBase->execute_uid);

    char strJobId[32];

    char logFile[PATH_MAX+1];
    snprintf(logFile, PATH_MAX, "%s/jobstatus.%s",
             tmpDirName, jobid_2longstring(jobBase->jobid, strJobId));

    FILE *fp;
    if ((fp = fopen(logFile, "w")) == NULL) {
        change_uid(guidManager);
        return -1;
    }

    unsigned int num = 0;
    struct record log;
    memset(&log, 0, sizeof(struct record));

    snprintf(log.version, MAXVERSIONLEN, "%d", QP_XDR_VERSION);
    log.type = RECORD_TYPE_JM_JOB_STATUS;
    log.log_time = time(NULL);

    num ++; /* jobid */

    if (0 < jobBase->status) {
        num ++;
    }

    if (0 < (jobBase->reasons & (~SUSPEND_REASON_JMSTART))) {
        num ++;
    }

    if (0 < jobBase->subreasons) {
        num ++;
    }

    if (0 < jobBase->act_pid) {
        num ++;
    }

    if (0 < jobBase->act_signal) {
        num ++;
    }

    if (0 < jobBase->submit->chkpnt_duration) {
        num ++;
    }

    if (0 < job->actcmd_state) {
        num ++;
    }

    if (0 < job->susp_reason) {
        num ++;
    }

    if (0 < job->susp_subreason) {
        num ++;
    }

    if (0 < jobBase->job_pid) {
        num ++;
    }

    if (0 < jobBase->job_pgid) {
        num ++;
    }

    log.KVs = (key_value_t *)calloc(num, sizeof(key_value_t));
    if (NULL == log.KVs) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", num*sizeof(key_value_t));
        return -1;
    }

    log.numKVs = 0;

    log.KVs[log.numKVs].key = RECORD_KEY_JOBID;
    log.KVs[log.numKVs].value = (char *)calloc(32, sizeof(char));
    if (NULL == log.KVs[log.numKVs].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 32);
        relaunch();
    }
    jobid_2longstring(jobBase->jobid, log.KVs[log.numKVs].value);
    log.numKVs ++;

    if (0 < jobBase->status) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_STATUS;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->status);
        log.numKVs ++;
    }

    if (0 < (jobBase->reasons & (~SUSPEND_REASON_JMSTART))) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_REASON;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d",
                 jobBase->reasons & (~SUSPEND_REASON_JMSTART));
        log.numKVs ++;
    }

    if (0 < jobBase->subreasons) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_SUBREASON;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->subreasons);
        log.numKVs ++;
    }

    if (0 < jobBase->act_pid) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_ACTPID;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->act_pid);
        log.numKVs ++;
    }

    if (0 < jobBase->act_signal) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_ACTVALUE;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->act_signal);
        log.numKVs ++;
    }

    if (0 < jobBase->submit->chkpnt_duration) {
        log.KVs[log.numKVs].key = RECORD_KEY_CHECKPOINT_PERIOD;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->submit->chkpnt_duration);
        log.numKVs ++;
    }

    if (0 < job->actcmd_state) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_ACTSTATUS;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", job->actcmd_state);
        log.numKVs ++;
    }

    if (0 < job->susp_reason) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_ACTREASON;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", job->susp_reason);
        log.numKVs ++;
    }

    if (0 < job->susp_subreason) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_ACTSUBREASON;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", job->susp_subreason);
        log.numKVs ++;
    }

    if (0 < jobBase->job_pid) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_PID;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->job_pid);
        log.numKVs ++;
    }

    if (0 < jobBase->job_pgid) {
        log.KVs[log.numKVs].key = RECORD_KEY_JOB_PGID;
        log.KVs[log.numKVs].value = (char *)calloc(12, sizeof(char));
        if (NULL == log.KVs[log.numKVs].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(log.KVs[log.numKVs].value, 12, "%d", jobBase->job_pgid);
        log.numKVs ++;
    }

    if (write_record(fp, &log) == -1) {
        eno = errno;
        fclose(fp);
        fp = NULL;
        errno = eno;
        change_uid(guidManager);
        return -1;
    }

    fclose(fp);
    change_uid(guidManager);

    return 0;
} // end function jmlog_newstatus

/**
 * @brief       Recover job status during jm recovering.
 *
 * @param[in|out]   job      #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
int
jmread_jobstatus(struct jobEntry *job)
{
    unsigned int lineNum = 0;
    struct jobInfo *jobBase = &(job->job_base);

    char tmpDirName[PATH_MAX+1];
    get_tmpdir_4job(tmpDirName, job->job_base.jobid, job->job_base.execute_uid);

    char strJobId[32];

    char logFile[PATH_MAX+1];
    snprintf(logFile, PATH_MAX, "%s/jobstatus.%s",
             tmpDirName, jobid_2longstring(jobBase->jobid, strJobId));

    FILE *fp;
    if ((fp = fopen(logFile, "r")) == NULL) {

        int jobid32, jobid, idx;
        jobid = ARRAY_JOBID(jobBase->jobid);
        idx = ARRAY_INDEX(jobBase->jobid);
        jobid32 = (unsigned int)idx << 20 | jobid;
        snprintf(logFile, PATH_MAX, "%s/jobstatus.%d", tmpDirName, jobid32);
        if((fp = fopen(logFile, "r")) == NULL) {
            return -1;
        }
    }

    struct record *statusRecord;
    statusRecord = read_file_line(fp, &lineNum);
    if (NULL == statusRecord || statusRecord->type != RECORD_TYPE_JM_JOB_STATUS) {

        fclose(fp);
        return -1;
    }

    unsigned int i;
    for (i=0; i<statusRecord->numKVs; i++) {
        switch (statusRecord->KVs[i].key) {
        case RECORD_KEY_JOBID:
            jobBase->jobid = atoll(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_STATUS:
            jobBase->status = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_REASON:
            jobBase->reasons = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_SUBREASON:
            jobBase->subreasons = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_ACTPID:
            jobBase->act_pid = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_ACTVALUE:
            jobBase->act_signal = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_CHECKPOINT_PERIOD:
            jobBase->submit->chkpnt_duration = atol(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_ACTSTATUS:
            job->actcmd_state = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_ACTREASON:
            job->susp_reason = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_ACTSUBREASON:
            job->susp_subreason = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_PID:
            jobBase->job_pid = atoi(statusRecord->KVs[i].value);
            break;
        case RECORD_KEY_JOB_PGID:
            jobBase->job_pgid = atoi(statusRecord->KVs[i].value);
            break;
        }
    }

    fclose(fp);

    return 0;
} // end function jmread_jobstatus

/**
 * @brief       Set execute user before job starting.
 *
 * @param[in|out]   job      #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
int
set_execute_user(struct jobEntry *job)
{
    struct passwd *pw;
    char strJobId[32];

    pw = get_user_byname(job->job_base.submit->user);
    if (NULL == pw) {
        if (errno == 0) {
            log_message(ERR, NOCLASS, "%s: No execution user <%s> found for job <%s>.",
                        __func__, job->job_base.submit->user,
                        jobid_2string(job->job_base.jobid, strJobId));
        } else {
            log_message(ERR, NOCLASS, "%s: No execution user <%s> found for job <%s> due to %m.",
                        __func__, job->job_base.submit->user,
                        jobid_2string(job->job_base.jobid, strJobId));
        }
        return -1;
    }

    job->execute_gid = pw->pw_gid;
    job->job_base.execute_uid = pw->pw_uid;

    FREEUP(job->job_base.execute_user);
    job->job_base.execute_user = copy_string(job->job_base.submit->user);

    return 0;
} // end function set_execute_user

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

static int
get_pidpgid(char *message, pid_t *pid, pid_t *pgid)
{
    char *strPidMsg, *strPGidMsg;

    if ((strPidMsg = strstr(message, "pid=")) != NULL
        && (strPGidMsg = strstr(message, "pgid=")) != NULL) {

        *pid = atoi(strPidMsg + strlen("pid="));
        *pgid = atoi(strPGidMsg + strlen("pgid="));

        return 0;
    }

    return -1;
} // end function get_pidpgid

/**
 * @brief       Call erestart to restart the checkpoint job.
 *
 * @param[in]   job          #1: job entry to be executed
 *
 * @note        fork child to call erestart
 */
static void
execute_restart_job(struct jobEntry *job)
{
    struct jobInfo*jobBase = &(job->job_base);
    char *rargv[MAX_ARGS];
    int i;
    int fds[2];
    char chkDir[PATH_MAX+1],
        chkpntDir[PATH_MAX+1],
        restartDir[PATH_MAX+1],
        oldChkpntDir[PATH_MAX+1];
    char oldJobId[32], newJobId[32];

    Signal_(SIGTERM, SIG_IGN);
    Signal_(SIGINT, SIG_IGN);
    Signal_(SIGUSR1, SIG_IGN);
    Signal_(SIGUSR2, SIG_IGN);

    char strJobId[32];

    char *strDir;
    if (((strDir = strrchr(jobBase->submit->chkpnt_dir, '/')) != NULL) && (islong(strDir+1))) {

        snprintf(chkpntDir, PATH_MAX, "%s", jobBase->submit->chkpnt_dir);
        *strDir = '\0';
        snprintf(chkDir, PATH_MAX, "%s", jobBase->submit->chkpnt_dir);
        snprintf(oldJobId, 32, "%s", strDir+1);
        snprintf(newJobId, 32, "%s", jobid_2string(jobBase->jobid, strJobId));
        *strDir = '/';
    } else {

        log_message(ERR, NOCLASS, "%s: %s is not a valid check point directory.",
                    __func__, jobBase->submit->chkpnt_dir);
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_CHKPNT_DIR, job);
        exit (-1);
    }
    
    snprintf(restartDir, PATH_MAX, "%s/%s", chkDir, newJobId);
    snprintf(oldChkpntDir, PATH_MAX, "%s/%s", chkDir, oldJobId);

    if (pipe(fds) == -1) {

        log_message(ERR, NOCLASS, "%s: pipe() failed for job[%s]: %m",
                    __func__, jobid_2string(jobBase->jobid, strJobId));
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_SYSCALL, job);
    }

    pid_t pid;
    pid = fork();
    if (0 > pid) {

        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M,
                    __func__, jobid_2string(jobBase->jobid, strJobId), "fork");
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_FORK, job);
    }

    if (0 < pid) {

        close(fds[1]);

        pid_t erestartPid, erestartPGid;
        erestartPid = pid;
        erestartPGid = jobBase->job_pgid;

        char strRestartMsg[MAXMESSAGELENGTH];
        char *sp = strRestartMsg;
        int len = sizeof(strRestartMsg);

        errno = 0;
        strRestartMsg[0] = '\0';

    again:
        while (((i = read(fds[0], sp, MAXMESSAGELENGTH)) < 0)
               && (errno == EINTR));

        if ((i  > 0) && (errno == EINTR)) {
            sp += i;
            goto again;
        }
        if (i  > 0) {
            sp += i;
        }
        *sp = '\0';

        if ((strlen(strRestartMsg) == 0)
            || (get_pidpgid(strRestartMsg, &erestartPid, &erestartPGid) != -1)) {

            jobBase->job_pgid = erestartPGid;
            childjm_set_jobstatus(JOB_STATUS_RUN, 0, job);
        } else {
            log_message(ERR, NOCLASS, "%s: Unable to restart job <%s>: (%s)",
                        __func__, jobid_2string(jobBase->jobid, strJobId), strRestartMsg);
            childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_RESTART_JOBPID, job);
        }

        sp = strRestartMsg;
        *sp = '\0';
        while ((i = read(fds[0], sp, len)) > 0) {
            sp += i;
            len -= i;
        }
        *sp = '\0';

        if (0 < strlen(strRestartMsg)) {
            log_message(ERR, NOCLASS, "%s: %s", __func__, strRestartMsg);
        }

        STATUS_WAIT_T status;
        while ((pid = waitpid(pid, &status, 0)) < 0 && errno == EINTR);

        if (pid < 0) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_D_M,
                        __func__, jobid_2string(jobBase->jobid, strJobId),
                        "waitpid", jobBase->job_pgid);
            rename(restartDir, oldChkpntDir);
            exit(-1);
        }

        exit(WEXITSTATUS(status));
    }

    i = 0;
    rargv[i++] = (char *)"erestart";

    rargv[i++] = (char *)"--method";
    rargv[i++] = jobBase->submit->chkpnt_method;

    rargv[i++] = restartDir;
    rargv[i] = NULL;

    {
        char buf[12];
        snprintf(buf, 12, "%d", dup(2));
        gf_setenv("QP_STDERR_FD", buf);
    }
    close(fds[0]);
    dup2(fds[1], 2);
    close(fds[1]);

    char strPid[20], strPgid[20];
    snprintf(strPid, 12, "%d", jobBase->job_pid);
    gf_setenv("QP_RESTART_PID", strPid);

    snprintf(strPgid, 12, "%d", jobBase->restart_pid);
    gf_setenv("QP_RESTART_PGID", strPgid);

    char *strExecRestart;
    char strChkpntDir[PATH_MAX+1];
    memset(strChkpntDir, 0, PATH_MAX+1);
    strExecRestart = get_chkpnt_dir("/erestart", strChkpntDir);
    if (NULL == strExecRestart) {
        exit(-1);
    }

    execv(strExecRestart, rargv);
    log_message(ERR, NOCLASS,STR_FUNC_S_FAIL_M, __func__, "execv", "erestart");

    exit(-1);
} // end function execute_restart_job

/**
 * @brief       colloect queue level and user level job pre script status.
 *
 * @param[in]   job          #1: job
 * @param[in]   pid          #2: pid of the script
 * @param[in]   context      #3: caller funcation name
 */
static void
collect_precmd_status(struct jobEntry *job, pid_t pid, const char *context)
{
    int id;
    STATUS_WAIT_T status;
    struct rusage rusage;
    char strJobId[32];
    while ((id = wait3(&status, 0, &rusage)) != pid) {

        if (id < 0) {
            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_D_M, __func__,
                        jobid_2string(job->job_base.jobid, strJobId), "wait3", pid);
            childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_SYSCALL, job);
        }

        log_message(DEBUG, EXECUTE, "%s: %s: wait3() got %d, not pid %d for job <%s>",
	            __func__, context, id, pid, jobid_2string(job->job_base.jobid, strJobId));
    }

    if (!JOB_WSTATUS(status)) {
        return;
    }

    log_message(DEBUG, EXECUTE, "%s: %s: job <%s> status <%d>",
                __func__, context, jobid_2string(job->job_base.jobid, strJobId), JOB_WSTATUS(status));

    job->w_status = JOB_WSTATUS(status);

    double ru_utime;
    ru_utime = rusage.ru_utime.tv_sec + rusage.ru_utime.tv_usec / 1000000.0;

    double ru_stime;
    ru_stime = rusage.ru_stime.tv_sec + rusage.ru_stime.tv_usec / 1000000.0;

    job->cpu_time = ru_utime + ru_stime;

    if ((WIFEXITED(status) && WEXITSTATUS(status) == PRE_EXECUTION_ABORT)
        || WIFSIGNALED(status)) {

        childjm_set_jobstatus(JOB_STATUS_EXIT, 0, job);
    }

    if (0 == strcmp(context, "execute_user_precmd")) {
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_PRESCRIPT, job);
    } else {
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_QUEUE_PRESCRIPT, job);
    }

    return;
} // end function collect_precmd_status

/**
 * @brief       fork a child to execute queue specified job pre-script.
 *
 * @param[in]   job          #1: job
 */
static void
execute_queue_precmd(struct jobEntry *job)
{
    int i;

    if (job->job_base.prequeue_command == NULL || job->job_base.prequeue_command[0] == '\0') {
        return;
    }

    Signal_(SIGCHLD, SIG_DFL);

    sigset_t new_mask, old_mask;
    sigemptyset(&new_mask);
    sigaddset(&new_mask, SIGCHLD);
    sigprocmask(SIG_SETMASK, &new_mask, &old_mask);

    char strJobId[32];
    jobid_2string(job->job_base.jobid, strJobId);

    pid_t pid;
    pid = fork();
    if (0 > pid) {

        log_message(ERR, NOCLASS, "%s: fork() failed during execute queue pre script for job <%s> due to %m",
                    __func__, strJobId);

        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_QUEUE_PRESCRIPT, job);
        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return;
    }

    if (0 == pid) {  //child, to execute queue pre script.
        sigset_t newmask;
        char *myargv[5];

        if (0 > chdir("/tmp")) {
            log_message(ERR, NOCLASS, "%s: failed to chkdir(/tmp) for job <%s> due to %m",
                        __func__, strJobId);
        }

        if (setuid_4prepost_cmd(job) != 0) {
            log_message(ERR, NOCLASS, "%s: setuid_4prepost_cmd() failed for job <%s> 's pre script.",
	                __func__, strJobId);
            exit(-1);
        }

        for (i = 1; i < NSIG; i++) {
            Signal_(i, SIG_DFL);
        }

        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);

        alarm(0);

        gf_setenv("PATH", (char *)"/bin:/usr/bin:/sbin:/usr/sbin");

        i = open("/dev/null", O_RDWR, 0);
        dup2(i, 0);
        dup2(i, 1);
        dup2(i, 2);

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
                 baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
        open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                 baseParams[BASE_JM_TIME_DEBUG].int_value);

        myargv[0] = (char *)"/bin/sh";
        myargv[1] = (char *)"-c";
        myargv[2] = job->job_base.prequeue_command;
        myargv[3] = NULL;

        execvp ("/bin/sh", myargv);

        log_message(ERR, NOCLASS, "%s: execute queue pre script \"%s\" failed for job <%s> due to %m",
                    __func__, job->job_base.prequeue_command, strJobId);
        exit (-1);
    }

    //parent process, wait for the pre-exec process finishing and collect the exit status
    collect_precmd_status(job, pid, __func__);
    job->execute_flag |= JOB_EXEC_QPRE_OK;

    sigprocmask(SIG_SETMASK, &old_mask, NULL);

    return;
} // end function execute_queue_precmd

/**
 * @brief       fork a child to execute user specified job pre-script.
 *
 * @param[in]   job          #1: job
 *
 * @note        user pre-script is executed before job is started.
 */
static void
execute_user_precmd(struct jobEntry *job)
{
    Signal_(SIGCHLD, SIG_DFL);

    sigset_t new_mask, old_mask;
    sigemptyset(&new_mask);
    sigaddset(&new_mask, SIGCHLD);
    sigprocmask(SIG_SETMASK, &new_mask, &old_mask);

    pid_t pid;
    struct jobInfo *jobBase = &job->job_base;
    char strJobId[32];
    if ((pid = fork()) < 0) {

        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "fork");
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_FORK, job);
        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return;
    }

    if (pid == 0) {

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
                 baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
        open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                 baseParams[BASE_JM_TIME_DEBUG].int_value);

        if (getuid() == guidManager) {

            change_uid(guidManager);

            if (0 > setuid(jobBase->execute_uid)) {
                log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                            jobid_2string(jobBase->jobid, strJobId), "setuid");
            }
        }

        int i;
        for (i = 1; i < NSIG; i++)
            Signal_(i, SIG_DFL);

        Signal_(SIGHUP, SIG_IGN);

        sigset_t newmask;
        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);
        alarm(0);

        execl("/bin/sh", "/bin/sh", "-c", jobBase->submit->prejob_command, NULL);
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "execl");
        log_message(ERR, NOCLASS, jobBase->submit->prejob_command);

        exit(-1);
    }

    collect_precmd_status(job, pid, __func__);
    jobBase->status &= ~JOB_STATUS_PREJOB;
    sigprocmask(SIG_SETMASK, &old_mask, NULL);

    return;
} // end function execute_user_precmd

/**
 * @brief       fork a child to execute user specified job post-script.
 *
 * @param[in]   job          #1: job
 *
 * @note        job post-script is executed after job is finished.
 */
static void
execute_user_postcmd(struct jobEntry *job)
{    
    Signal_(SIGCHLD, SIG_DFL);

    sigset_t new_mask, old_mask;
    sigemptyset(&new_mask);
    sigaddset(&new_mask, SIGCHLD);
    sigprocmask(SIG_SETMASK, &new_mask, &old_mask);

    struct jobInfo *jobBase = &job->job_base;
    char strJobId[32];

    pid_t pid;
    pid = fork();
    if (0 > pid) {

        change_uid(guidManager);
        log_message(ERR, NOCLASS, "%s: fork() failed for job %d: %m",
                    __func__, jobBase->jobid);
        change_uid(job->job_base.execute_uid);
        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return;
    }

    if (0 == pid) { // child, to execute user post script.

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
                 baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
        open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                 baseParams[BASE_JM_TIME_DEBUG].int_value);

        if (getuid() == guidManager) {

            change_uid(guidManager);

            if (0 > setuid(jobBase->execute_uid)) {
                log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                            jobid_2string(jobBase->jobid, strJobId), "setuid");
            }
        }

        int i;
        for (i = 1; i < NSIG; i++)
            Signal_(i, SIG_DFL);

        Signal_(SIGHUP, SIG_IGN);

        sigset_t newmask;
        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);
        alarm(0);

        execl("/bin/sh", "/bin/sh", "-c", jobBase->submit->postjob_command, NULL);
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "execl");
        log_message(ERR, NOCLASS, jobBase->submit->postjob_command);

        exit(-1);
    }

    if (waitpid(pid, NULL, 0) < 0) {

        change_uid(guidManager);
        log_message(ERR, NOCLASS, "%s: waitpid(%d) failed for user post for job <%d>: %m",
              __func__, pid, job->job_base.jobid);
        change_uid(job->job_base.execute_uid);
        sigprocmask(SIG_SETMASK, &old_mask, NULL);

        return;
    }
    sigprocmask(SIG_SETMASK, &old_mask, NULL);

    return;
} // end function execute_user_postcmd

/**
 * @brief       Set runtime resource limits for the job.
 *
 * @param[in]   jobbase      #1: job information
 *
 * @note        Set limit to avoid job using too much resource.
 */
static void
set_job_limits(struct jobInfo *jobbase)
{
    struct rlimit rlimit;
    char strJobId[32];

#ifdef  RLIMIT_CPU
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_CPU], &rlimit, RESOURCE_LIMIT_CPU);
    if (setrlimit(RLIMIT_CPU, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_CPU) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (long)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif


#ifdef  RLIMIT_FSIZE
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_FSIZE], &rlimit, RESOURCE_LIMIT_FSIZE);
    if (setrlimit(RLIMIT_FSIZE, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_FSIZE) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (long)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif

#ifdef RLIMIT_DATA
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_DATA], &rlimit, RESOURCE_LIMIT_DATA);
    if (setrlimit(RLIMIT_DATA, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_DATA) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (long)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif

#ifdef  RLIMIT_STACK
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_STACK], &rlimit, RESOURCE_LIMIT_STACK);
    if (setrlimit(RLIMIT_STACK, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_STACK) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (long)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif

#ifdef RLIMIT_CORE
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_CORE], &rlimit, RESOURCE_LIMIT_CORE);
    if (setrlimit(RLIMIT_CORE, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_CORE) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (long)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif

#ifdef RLIMIT_RSS
    set_resource_limit(&jobbase->resource_limits[RESOURCE_LIMIT_RSS], &rlimit, RESOURCE_LIMIT_RSS);
    if (setrlimit(RLIMIT_RSS, &rlimit) < 0) {
        log_message(ERR, NOCLASS, "%s: setrlimit(RLIMIT_RSS) failed for job <%s>: %m: soft %ld hard %ld",
                    __func__, jobid_2string(jobbase->jobid, strJobId),
                    (int)rlimit.rlim_cur, rlimit.rlim_max);
    }
#endif

    return;
} // end function set_job_limits

/**
 * @brief       Start executing the user job.
 *
 * @param[in]   job          #1: job
 *
 * @note        set pgid, env and dir before job starting.
 */
static void
execute_job(struct jobEntry *job)
{
    int i;
    sigset_t newmask;

    struct jobInfo *jobBase = &(job->job_base);
    jobBase->job_pid = getpid();
    jobBase->job_pgid = jobBase->job_pid;

    char strJobId[32];
    if (setpgid(0, jobBase->job_pgid) < 0) {

        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                    jobid_2string(jobBase->jobid, strJobId), "setpgid");

        jobBase->job_pgid = -1;
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_SYSCALL, job);
    }

    if (set_env_4job(job) < 0) {

        log_message(DEBUG, EXECUTE, "%s: set_env_4job() failed for job <%s>",
                    __func__, jobid_2string(jobBase->jobid, strJobId));

        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_EXECSCRIPT, job);
    }

    if (setup_gpu_option(job)) {
        log_message(ERR, NOCLASS, "%s: failed to control gpu for job[%s]",
                    __func__, jobid_2string(jobBase->jobid, strJobId));
        exit(-1);
    }

    umask(jobBase->submit->umask);

    execute_queue_precmd(job);

    errno = 0;
    if (nice(NICE_MIDDLE) == -1 && errno != 0) {
        log_message(ERR, NOCLASS, "%s: nice(%d) failed %m", __func__, NICE_MIDDLE);
    }

    if (nice(jobBase->nice) == -1 && errno != 0) {
        log_message(ERR, NOCLASS, "%s: nice(%d) failed %m", __func__, jobBase->nice);
    }

    set_job_limits(jobBase);

    //configuration file specifies whether to use root
    if (!((jobBase->submit->options & SUBOPT_RESTART)
          && baseParams[QP_CHKPNT_ROOT].string_value != NULL
          && strcmp(baseParams[QP_CHKPNT_ROOT].string_value, "Y") == 0)) { 

        if (set_userid_4job(jobBase->jobid, jobBase->execute_user,
                            job->execute_gid, jobBase->execute_uid) < 0) {
            childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_SYSCALL, job);
        }
    }

    if (setup_path_4job(job, &jobBase->job_file) < 0) {
        childjm_set_jobstatus(JOB_STATUS_PEND, REASON_JM_SETUP_DIR, job);
    }

    for (i = 1; i < NSIG; i++) {
        Signal_(i, SIG_DFL);
    }

    Signal_(SIGHUP, SIG_IGN);

    sigemptyset(&newmask);
    sigprocmask(SIG_SETMASK, &newmask, NULL);

    /* unblock all signals */
    alarm(0);

    char strJobTmpDir[PATH_MAX+1];
    get_tmpdir_4job(strJobTmpDir, job->job_base.jobid, job->job_base.execute_uid);
    if ('\0' != strJobTmpDir[0]) {
        gf_setenv("TMPDIR", strJobTmpDir);
    }

    if (NULL != jobBase->submit->prejob_command && '\0' != jobBase->submit->prejob_command[0]) {
        execute_user_precmd(job);
    }

    childjm_set_jobstatus(JOB_STATUS_RUN, 0, job);

    if (jobBase->submit->options & SUBOPT_RESTART) {
        execute_restart_job(job);
    }

    if (!(jobBase->submit->options & SUBOPT_RESTART)) {
        char *jobArgv[4];

        if (NULL == jobBase->submit->login_shell || '\0' == jobBase->submit->login_shell[0]) {

            jobArgv[0] = jobBase->submit->script;
            jobArgv[1] = NULL;

            char **execArgv;
            execArgv = set_startjob_arguments(jobBase->submit->from_node, jobArgv);
            if (NULL == execArgv) {
                exit(-1);
            }

            execvp(execArgv[0], execArgv);

            log_message(ERR, NOCLASS, "%s: unable to execute jobfile %s for job %s due to %m.",
                        __func__, execArgv[0], jobid_2string(jobBase->jobid, strJobId));
        } else {
            char *shellPath = NULL;

            shellPath = jobBase->submit->login_shell;
            if (NULL == shellPath) {
                shellPath = (char *)"/bin/sh";
            }

            putenv((char *)"PATH=/bin:/usr/bin:/local/bin:/usr/local/bin");

            if (0 > chdir(jobBase->execute_home)) {
                log_message(ERR, NOCLASS, "%s: failed to chkdir(%s) for job <%s> due to %m",
                            __func__, jobBase->execute_home, jobid_2string(jobBase->jobid, strJobId));
            }

            gf_setenv("SHELL", shellPath);

            jobArgv[0] = (char *)"-";
            jobArgv[1] = NULL;
            execv(shellPath, jobArgv);

            fprintf(stderr, "%s: Unable to execute login shell %s for job %s: %m\n",
                    __func__, shellPath, jobid_2string(jobBase->jobid, strJobId));
        }
    }

    exit(-1);
} // end function execute_job

/**
 * @brief       Set environment variables for the job.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, run_exec failure
 */
static int
set_env_4job(struct jobEntry *job)
{
    struct jobInfo *jobBase = &job->job_base;

    char strJobId[32];
    log_message(DEBUG, EXECUTE, "%s: job <%s> has %d environment variables",
                __func__, jobid_2string(jobBase->jobid, strJobId), jobBase->num_environment);

    char *tz, tzsave[256+1];
    if ((tz = getenv("TZ"))) {
        strncpy(tzsave, tz, 256);
    }

    static char **env;
    if (env != NULL) {
        FREEUP(env);
    }

    env = (char **)calloc(2, sizeof(char *));
    if (NULL == env) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 2*sizeof(char *));
        relaunch();
    }

    env[0] = NULL;
    environ = env;

    unsigned int i;
    for (i = 0; i < jobBase->num_environment; i++) {
        char *sp;

        if ((sp = strchr (jobBase->environment[i], '=')) == NULL) {
            continue;
        }

        if ((NULL != jobBase->submit->login_shell && jobBase->submit->login_shell[0] != '\0')
            && (strstr(jobBase->environment[i], "QP_ACCT_MAP") == NULL)
            && (strstr(jobBase->environment[i], "QP_UNIXGROUP") == NULL)
            && (strstr(jobBase->environment[i], "QP_JOB_STARTER") == NULL)
            && (strstr(jobBase->environment[i], "LOGNAME") == NULL)) {

            continue;
        }

        *sp = '\0';
        sp++;
        gf_setenv(jobBase->environment[i], sp);
        sp--;
        *sp = '=';
    }

    if (tz) {
        gf_setenv("TZ", tzsave);
    }

    char val[PATH_MAX+1];
    snprintf(val, PATH_MAX, "%d", ARRAY_JOBID(jobBase->jobid));
    gf_setenv("QP_JOBID", val);

    snprintf(val, PATH_MAX, "%d", ARRAY_INDEX(jobBase->jobid));
    gf_setenv ("QP_JOBINDEX", val);

    gf_setenv("QP_JOBFILENAME", jobBase->submit->script);

    char shellFile[PATH_MAX+1];
    if (strlen(jobBase->submit->chkpnt_dir) == 0) { 
        strncpy(shellFile,jobBase->submit->script, PATH_MAX);
    } else {
        snprintf(shellFile, PATH_MAX, "%s/%s",
                 jobBase->submit->chkpnt_dir, jobBase->submit->script);
    }    

    if (strlen(jobBase->submit->chkpnt_dir) > 0) { 
        char chkDir[PATH_MAX+1];
        char cwd[PATH_MAX+1];
        char *strDir;

        if (((strDir = strrchr(jobBase->submit->chkpnt_dir, '/')) != NULL) && (islong(strDir+1))) {
            if (jobBase->submit->chkpnt_dir[0] == '/') {

                if(jobBase->submit->options & SUBOPT_RESTART) {
                    *strDir = '\0';
                    snprintf(chkDir, PATH_MAX, "%s/%s", jobBase->submit->chkpnt_dir,
                             jobid_2string(jobBase->jobid, strJobId));
                    *strDir = '/'; 
                } else {
                    snprintf(chkDir, PATH_MAX, "%s",jobBase->submit->chkpnt_dir);
                }    
            } else {

                if (jobBase->submit->cwd[0] == '/') {
                    strncpy(cwd, jobBase->submit->cwd, PATH_MAX);
                } else {
                    if (jobBase->submit->cwd[0] == '\0') {
                        strncpy(cwd, jobBase->submit->home_dir, PATH_MAX);
                    } else {
                        snprintf(cwd, PATH_MAX, "%s/%s",
                                 jobBase->submit->home_dir, jobBase->submit->cwd);
                    }
                }

                if (jobBase->submit->options & SUBOPT_RESTART) {
                    *strDir = '\0';
                    snprintf(chkDir, PATH_MAX, "%s/%s/%s",
                             cwd, jobBase->submit->chkpnt_dir,
                             jobid_2string(jobBase->jobid, strJobId));
                    *strDir = '/';
                } else {
                    snprintf(chkDir, PATH_MAX, "%s", jobBase->submit->chkpnt_dir);
                }
            }

            gf_setenv("QP_CHKPNT_DIR", chkDir);
            snprintf(shellFile, PATH_MAX, "%s/%s", chkDir, jobBase->submit->script);
        } else {
            snprintf(shellFile, PATH_MAX, "%s/%s",
                     jobBase->submit->chkpnt_dir, jobBase->submit->script);
            log_message(ERR, NOCLASS, "Failed to set QP_CHKPNT_DIR, error in the chkpnt directory: %s",
                        jobBase->submit->chkpnt_dir);
        }

        snprintf(val, 12, "%d", jobBase->submit->chkpnt_duration);
        gf_setenv("QP_CHKPNT_PERIOD", val);
    }

    gf_setenv("QP_CHKFILENAME", shellFile);

    snprintf(val, PATH_MAX, "trap # %d %d %d %d %d", SIGTERM, SIGUSR1, SIGUSR2, SIGINT, SIGHUP);
    gf_setenv("QP_TRAPSIGS", val);

#define MAX_QP_HOSTS_LEN 4096
    if (jobBase->number_node <= (MAX_QP_HOSTS_LEN/2)) {

        char *hosts;
        hosts = (char *)calloc(jobBase->number_node * (MAXHOSTNAMELEN+20), sizeof(char));
        if (NULL == hosts) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", jobBase->number_node * (MAXHOSTNAMELEN+20));
            relaunch();
        }

        char *ompiHosts;
        ompiHosts = (char *)calloc(jobBase->number_node * (MAXHOSTNAMELEN+20), sizeof(char));
        if (NULL == ompiHosts) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", jobBase->number_node * (MAXHOSTNAMELEN+20));
            relaunch();
        }

        for (i=0; i<jobBase->number_node-1; i++) {
            snprintf(hosts+strlen(hosts), MAX_QP_HOSTS_LEN, "%s:%d",
                     jobBase->node_execinfo[i].host_name, jobBase->node_execinfo[i].num_slot);
            strcat(hosts, " ");

            snprintf(ompiHosts+strlen(ompiHosts), MAX_QP_HOSTS_LEN, "%s slots=%d",
                     jobBase->node_execinfo[i].host_name, jobBase->node_execinfo[i].num_slot);
            strcat(ompiHosts, "\n");
        }
        snprintf(hosts+strlen(hosts), MAX_QP_HOSTS_LEN, "%s:%d",
                 jobBase->node_execinfo[i].host_name, jobBase->node_execinfo[i].num_slot);
        snprintf(ompiHosts+strlen(ompiHosts), MAX_QP_HOSTS_LEN, "%s slots=%d",
                 jobBase->node_execinfo[i].host_name, jobBase->node_execinfo[i].num_slot);

        if (strlen(hosts) < MAX_QP_HOSTS_LEN) {
            gf_setenv("QP_HOSTS", hosts);
        }
        FREEUP(hosts);

        if (strlen(ompiHosts) < MAX_QP_HOSTS_LEN) {
            gf_setenv("QP_HOSTS_OMPI", ompiHosts);
        }
        FREEUP(ompiHosts);
    }

    /* For compatibility, some scripts parse unfolded hosts environment variables.
     */
    if (jobBase->number_node <= (MAX_QP_HOSTS_LEN /2))  {
        char *hosts;
        int nhosts = 0;
        int len;
        int pos = 0;

        for (i = 0; i < jobBase->number_node; i++) {
            nhosts += jobBase->node_execinfo[i].num_slot;
        }
        len = nhosts * MAXHOSTNAMELEN;
        hosts = (char *)calloc(len+1, sizeof(char));
        if (NULL == hosts) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", jobBase->number_node * (MAXHOSTNAMELEN+20));
            relaunch();
        }

        for (i=0; i<jobBase->number_node; i++) {
            unsigned int j;
            for (j = 0; j < jobBase->node_execinfo[i].num_slot; j++) {
                pos += snprintf(hosts + pos, len - pos, "%s ", jobBase->node_execinfo[i].host_name);
            }
        }

        hosts[strlen(hosts) - 1] = '\0';

        if (strlen(hosts) < MAX_QP_HOSTS_LEN) {
            gf_setenv("QP_HOSTS_COMPAT", hosts);
        }

        free(hosts);
    }

    if (NULL != jobBase->submit->output_file && '\0' != jobBase->submit->output_file[0]) {
        gf_setenv("QP_OUTPUTFILE", jobBase->submit->output_file);
    }

    if (NULL != jobBase->submit->error_file && '\0' != jobBase->submit->error_file[0]) {
        snprintf(val, PATH_MAX, "%s", jobBase->submit->error_file);
        gf_setenv("QP_ERRORFILE", val);
    }

    if (NULL != jobBase->submit->queue && '\0' != jobBase->submit->queue[0]) {
        gf_setenv("QP_QUEUE", jobBase->submit->queue);
    }

    if (NULL != jobBase->submit->job_name && '\0' != jobBase->submit->job_name[0]) {
        gf_setenv("QP_JOBNAME", jobBase->submit->job_name);
    }

    if (NULL != jobBase->execute_user && '\0' != jobBase->execute_user[0]) {
        gf_setenv("QPUSER", jobBase->execute_user);
        gf_setenv("USER", jobBase->execute_user);
    }

    snprintf(val, 12, "%d", PRE_EXECUTION_ABORT);
    gf_setenv("QP_EXIT_PRE_ABORT", val);

    if (jobBase->requeue_exitcode && jobBase->requeue_exitcode[0] != '\0') {
        gf_setenv("QP_EXIT_REQUEUE", jobBase->requeue_exitcode);
    }

    if ((getenv("WINDIR") != NULL) || (getenv("windir") != NULL)) {
        char tmppath[PATH_MAX+1];

        snprintf(tmppath, PATH_MAX, "/bin:/usr/bin:/sbin:/usr/sbin");
        if (baseParams[BASE_BINDIR].string_value != NULL) {
            strncat(tmppath, ":", PATH_MAX);
            strncat(tmppath, baseParams[BASE_BINDIR].string_value, PATH_MAX);
        }
        gf_setenv("PATH",tmppath);
    } else {
        char *tmppath = NULL;
        char *envpath;
        int cc = TRUE;

        if (NULL != baseParams[BASE_BINDIR].string_value) {
            unsigned int len;

            envpath = getenv("PATH");
            if (NULL != envpath) {

                len = strlen(baseParams[BASE_BINDIR].string_value);
                cc  = strncmp(envpath,baseParams[BASE_BINDIR].string_value,len);
                if (cc != 0)
                    len += strlen(envpath) + 2;
            } else {
                len = strlen(baseParams[BASE_BINDIR].string_value) +2;
            }

            if (cc != 0) {

                tmppath = (char *)calloc(len, sizeof(char));
                if (NULL == tmppath) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len);
                    relaunch();
                }

                strcpy(tmppath, baseParams[BASE_BINDIR].string_value);
                if (envpath != NULL) {
                    strcat(tmppath,":");
                    strcat(tmppath, envpath);
                }
                gf_setenv("PATH", tmppath);
                FREEUP(tmppath);
            } else {
                gf_setenv("PATH", envpath);
            }
        }
    }

    snprintf(val, 12, "%d", (int)getpid());
    gf_setenv("QP_JOBPID", val);

    if (gf_isfullpath(jobBase->submit->cwd)) {
        gf_setenv("QP_SUBCWD", jobBase->submit->cwd);
    } else {
        char cwd[PATH_MAX+1];

        snprintf(cwd, PATH_MAX, "%s/%s", jobBase->submit->home_dir, jobBase->submit->cwd);
        gf_setenv("QP_SUBCWD", cwd);
    }

    if (NULL != jobBase->submit->from_node && '\0' != jobBase->submit->from_node[0]) {
        gf_setenv("QP_SUBMIT_HOST", jobBase->submit->from_node);
    }

    gf_setenv("ENVDIR", baseParams[BASE_CONFDIR].string_value);

    if (baseParams[BASE_SERVERDIR].string_value != NULL) {
        gf_setenv("SERVERDIR", baseParams[BASE_SERVERDIR].string_value);
    }
    if (baseParams[BASE_BINDIR].string_value != NULL) {
        gf_setenv("BINDIR", baseParams[BASE_BINDIR].string_value);
    }
    if (baseParams[BASE_LIBDIR].string_value != NULL) {
        gf_setenv("LIBDIR", baseParams[BASE_LIBDIR].string_value);
    }

    snprintf(val, 12, "%d", job->w_status);
    gf_setenv("QP_JOBEXIT_STAT", val);

    snprintf(val, 12, "%d", jobBase->submit->max_job_slot);
    gf_setenv("QP_MAX_NUM_PROCESSORS", val);

    snprintf(val, 12, "%d", jobBase->number_node);
    gf_setenv("QP_NNODE", val);

    for (i=0; i<jobBase->number_node; i++) {
        snprintf(val, 12, "%d", jobBase->node_execinfo[i].num_slot);
        gf_setenv("QP_TASK_PERNODE", val);
        break;
    }

    if (NULL != jobBase->execute_user && '\0' != jobBase->execute_user[0]) {
        gf_setenv("QP_JOB_EXECUSER", jobBase->execute_user);
    }

    if (NULL != jobBase->requirement && '\0' != jobBase->requirement[0]) {
        gf_setenv("QP_EFFECTIVE_RSRCREQ", jobBase->requirement);
    }

    if (0 < jobBase->submit->thread_pertask) {
        snprintf(val, 12, "%d", jobBase->submit->thread_pertask);
        gf_setenv("OMP_NUM_THREADS", val);
    }

    int ret;
    ret = run_exec(EMPTY_STRING, jobBase->jobid, &jobBase->eexec, NULL);
    return ret;
} // end function set_env_4job

static void
remove_job_starter(char *line, char *jobstarter)
{
    static char *header=NULL, *tailer=NULL;
    static int header_len = 0;
    char *cmd;

    if (NULL == jobstarter || '\0' == jobstarter[0]) {
        return;
    }

    if (header == NULL) {
        header = jobstarter;
        if ((tailer = strstr(jobstarter, JOB_STARTER_KEYWORD)) != NULL) {

            *tailer = '\0';
            tailer += strlen (JOB_STARTER_KEYWORD);
        }
        header_len = strlen (header);
    }

    if (strncmp(line, header, header_len) != 0) {

        return;
    }

    if (tailer != NULL) {
        if ((cmd = strstr (line, tailer)) !=NULL) {

            *cmd = '\n';
            *(cmd+1) = '\0';
        }
    }

    cmd = line + strlen(header);
    while (*cmd == ' ' && *cmd != '\0') cmd++;

    memmove(line, cmd, (strlen(cmd)+1)*sizeof(char));

    return;
}

static int
is_linkfile(char *filename)
{
    if (NULL == filename || filename[0] == '\0') {
        return FALSE;
    }

    struct stat statBuf;
    if (0 == lstat(filename, &statBuf)) {

        if( S_ISLNK(statBuf.st_mode) ) {
            return TRUE;
        }
    }

    return FALSE;
}

static void
check_linkfile(struct jobEntry *job, bool_t *linkStdout, bool_t *linkStderr)
{
    *linkStdout = FALSE;
    *linkStderr = FALSE;

    if (!gboolStdoutDirect) {
        return;
    }

    char filename[PATH_MAX+1];

    struct jobInfo *jobBase = &job->job_base;
    if (NULL != jobBase->submit->output_file && '\0' != jobBase->submit->output_file[0]) {

        snprintf(filename, PATH_MAX, "%s.out", jobBase->submit->script);
        if (is_linkfile(filename)) {
            *linkStdout = TRUE;
        }
    }

    if (NULL != jobBase->submit->error_file && '\0' != jobBase->submit->error_file[0]) {

        snprintf(filename, PATH_MAX, "%s.err", jobBase->submit->script);
        if (is_linkfile(filename)) {
            *linkStderr = TRUE;
        }
    }

    log_message(DEBUG, EXECUTE, "%s: stdout is link file? %d stderr is link file? %d.",
                __func__, *linkStdout, *linkStderr);

    return;
}

static int
send_results(struct jobEntry *job)
{
    FILE *errout = NULL, *output = NULL, *notif = NULL;
    char line[MAXMESSAGELENGTH+1];
    char fileName[PATH_MAX+1];
    unsigned int k;
    unsigned int i;
    char ps[MAXMESSAGELENGTH*4+1];
    int ofIdx;
    char rcpMsg[MAXMESSAGELENGTH];
    STATUS_WAIT_T w_status;
    float cpuTime = job->cpu_time;
    char outputIsDirectory = FALSE;
    char errIsDirectory  = FALSE;
    char outputFileName[PATH_MAX+1];
    char errFileName[PATH_MAX+1];
    char jobIdFile[PATH_MAX+1];
    struct stat stb;
    char mailSizeStr[PATH_MAX+1];
    struct stat outfileStat;
    bool_t linkStdout;
    bool_t linkStderr;
    int errorOpeningOutputFile = FALSE;

    check_linkfile(job, &linkStdout, &linkStderr);

    struct jobInfo *jobBase = &job->job_base;
    outfileStat.st_size = -1;
    if (NULL == jobBase->submit->output_file || '\0' == jobBase->submit->output_file[0]) {

        snprintf(mailSizeStr, PATH_MAX, "%s.out", jobBase->submit->script);

        if (stat(mailSizeStr, &outfileStat) < 0) {
            log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "stat", mailSizeStr);
            outfileStat.st_size = -1;
        }
    }

    snprintf(mailSizeStr, PATH_MAX, "%jd", outfileStat.st_size);

    JOB_WSTATUS(w_status) = job->w_status;
    ps[0] = '\0';

    char *result;
    if (job->w_status)
        result = copy_string("Exited");
    else
        result = copy_string("Done");

    char withps = FALSE;
    int hasError=0;

    for (i = 0; i < jobBase->num_exec_files; i++) {

        if (jobBase->exec_files[i].direction & XF_OP_EXEC2SUB) {

            if (rcp_file(&job->job_base, jobBase->exec_files+i, jobBase->submit->from_node,
                         XF_OP_EXEC2SUB, rcpMsg) == 0) {

                if (rcpMsg[0] != '\0') {
                    snprintf(line, MAXMESSAGELENGTH,
                             "Copy file <%s> to <%s> on submission host <%s>: %s.\n",
                             jobBase->exec_files[i].out_file, jobBase->exec_files[i].in_file,
                             jobBase->submit->from_node, rcpMsg);
                    strncat(ps, line, MAXMESSAGELENGTH*4);
                    withps = TRUE;
                }
                continue;
            }

            snprintf(line, MAXMESSAGELENGTH, 
                     "Unable to copy file <%s> to <%s> on submission host <%s>: %s.\n",
                     jobBase->exec_files[i].out_file, jobBase->exec_files[i].in_file,
                     jobBase->submit->from_node, rcpMsg);
            hasError = -1;
            strncat(ps, line, MAXMESSAGELENGTH*4);
            withps = TRUE;
            return hasError;
        }
    }
    
    if (NULL != jobBase->submit->output_file && '\0' != jobBase->submit->output_file[0]) {
        char chr;
        int lastSlash = FALSE;
        int outDirOk = FALSE;


        chr = jobBase->submit->output_file[strlen(jobBase->submit->output_file) - 1];

        if (chr == '/' || chr == '\\' ) {
            outputIsDirectory = TRUE;
            lastSlash = TRUE;
        }

        if (stat(jobBase->submit->output_file, &stb) == 0 && S_ISDIR(stb.st_mode)) {
            outputIsDirectory = TRUE;
            outDirOk = TRUE;
        }

        if (!outputIsDirectory) {

            strncpy(outputFileName, jobBase->submit->output_file, PATH_MAX);
            outDirOk = TRUE;
        } else {
            strncpy(outputFileName, jobBase->submit->output_file, PATH_MAX);
            if (!outDirOk) {

                if (mkdir(outputFileName, 0700) == 0) {
                    outDirOk = TRUE;
                }
            }

            if (!lastSlash) {
                strncat(outputFileName, "/", PATH_MAX);
            }

            char strJobId[32];
            snprintf(jobIdFile, PATH_MAX, "%s.out", jobid_2longstring(jobBase->jobid, strJobId) );

            strncat(outputFileName, jobIdFile, PATH_MAX);
        }

        if (outDirOk) {
            output = fopen(outputFileName, "a");
        }

        if (NULL == output) {
            errorOpeningOutputFile = TRUE;

            if (!linkStdout) {
                snprintf(line, MAXMESSAGELENGTH,
                         "Fail to open output file %s: %m.\nOutput is stored in this mail.\n",
                         jobBase->submit->output_file);

            } else {
                snprintf(line, MAXMESSAGELENGTH, "Fail to open output file %s: %m.\n",
                         jobBase->submit->output_file);
            }

            strncat(ps, line, MAXMESSAGELENGTH*4);
            withps = TRUE;

            snprintf(mailSizeStr, PATH_MAX, "%s", "y");
            gf_setenv("QP_OUTPUT_TARGETFAILED", mailSizeStr);
            snprintf(mailSizeStr, PATH_MAX, "%s.out", jobBase->submit->script);
            if (stat(mailSizeStr, &outfileStat) < 0) {
                log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "stat", mailSizeStr);
                outfileStat.st_size = -1;
            }
            snprintf(mailSizeStr, PATH_MAX, "%jd", outfileStat.st_size);
            gf_setenv("QP_MAILSIZE", mailSizeStr);
        }
    } else {

        gf_setenv("QP_MAILSIZE", mailSizeStr);
    }

    if (!(jobBase->submit->options & SUBOPT_NOTIFY_END)) {
        notif = output;
    }

    int sendWarning = FALSE;
    if (notif == output) {
        sendWarning = TRUE;
    }

    if (NULL == notif) {
        FREEUP(result);
        return 0;
    }

    bool_t errFileError = FALSE;
    if (NULL != jobBase->submit->error_file && '\0' != jobBase->submit->error_file[0]) {
        char chr;
        int lastSlash = FALSE;
        int errDirOk = FALSE;

        chr = jobBase->submit->error_file[strlen(jobBase->submit->error_file) - 1];

        if (chr == '/' || chr == '\\' ) {
            errIsDirectory = TRUE;
            lastSlash = TRUE;
        }

        if (stat(jobBase->submit->error_file, &stb) == 0 && S_ISDIR(stb.st_mode)) {
            errIsDirectory = TRUE;
            errDirOk = TRUE;
        }

        if (!errIsDirectory) {

            strncpy(errFileName, jobBase->submit->error_file, PATH_MAX);
            errDirOk = TRUE;
        } else {

            strncpy(errFileName, jobBase->submit->error_file, PATH_MAX);
            if ( !errDirOk ) {

                if (mkdir(errFileName, 0700) == 0) {
                    errDirOk = TRUE;
                }
            }

            if (!lastSlash) {
                strncat(errFileName, "/", PATH_MAX);
            }

            char strJobId[32];
            snprintf(jobIdFile, PATH_MAX, "%s.err", jobid_2longstring(jobBase->jobid, strJobId) );

            strncat(errFileName, jobIdFile, PATH_MAX);
        }

        if ((errout = fopen(errFileName, "a")) == NULL) {
            snprintf(line, MAXMESSAGELENGTH, "Fail to open stderr file %s: %s.\nThe stderr output is included in this report.\n",
                    jobBase->submit->error_file, strerror(errno));
            strncat(ps, line, MAXMESSAGELENGTH*4);
            withps = TRUE;
            errout = notif;
            errFileError = TRUE;
        }
    } else {
        errout = output;
    }

    if (linkStdout && (output==notif) && !errorOpeningOutputFile ) {

        fputs("\n------------------------------------------------------------\n", notif);
    }

    fprintf(notif, "Sender: Quickpool System <%s@%s>\n",
            gptrMyCluster->admins[0], gptrMyHost->name);

    char strJobId[32];
    fprintf(notif, "Subject: Job %s: <%s> %s\n", jobid_2string(jobBase->jobid, strJobId),
            jobBase->submit->job_name, result);
    FREEUP(result);

    fprintf(notif, "\n");
    fprintf(notif, "Job <%s> was submitted from host <%s> by user <%s>.\n",
            jobBase->submit->job_name, jobBase->submit->from_node, jobBase->submit->user);

    fprintf(notif, "Job was executed on host(s) <%d*%s>, in queue <%s>, as user <%s>.\n",
            jobBase->node_execinfo[0].num_slot, jobBase->node_execinfo[0].host_name,
            jobBase->submit->queue, jobBase->execute_user);
    for (k=1; k < jobBase->number_node; k++) {
        fprintf(notif, "                            <%d*%s>\n",
                jobBase->node_execinfo[k].num_slot, jobBase->node_execinfo[k].host_name);
    }

    if (jobBase->execute_home[0] == '\0') {
        fprintf(notif, "Execution home directory was not known.\n");
    } else {
        fprintf(notif, "<%s> was used as the home directory.\n", jobBase->execute_home);
    }

    if (jobBase->execute_cwd[0] == '\0') {
        fprintf(notif, "Execution working directory was not known.\n");
    } else {
        fprintf(notif, "<%s> was used as the working directory.\n", jobBase->execute_cwd);
    }

    char timebuf[TIMESTRINGLEN];
    if (jobBase->start_time != 0) {
        ctime_2string(CTIME_FORMAT_DEFAULT, &jobBase->start_time, timebuf);
        fprintf(notif, "Started at %s\n", timebuf);
    } else {
        fprintf(notif, "Started at unknown time in the past\n");
    }

    time_t reportTime = time(NULL);
    fprintf(notif, "Results reported at %s",
            ctime_2string(CTIME_FORMAT_DEFAULT, &reportTime, timebuf));

    FILE *fp;
    if ((fp = fopen(jobBase->submit->script, "r")) == NULL) {
        fprintf(notif, "Cannot open your job script: %s\n", jobBase->submit->script);
    } else {
        while (fgets(line, PATH_MAX, fp) != NULL) {
            if (!strcmp(line, CMDSTART)) {
                break;
            }
        }

        if (line == NULL) {
            fprintf(notif, "Corrupted jobfile <%s>\n", jobBase->submit->script);
        } else {
            char *jobStarter;
            jobStarter=getenv("QP_JOB_STARTER");
            fputs("\n%s\nYour job looked like:", notif);
            fputs("\n------------------------------------------------------------\n", notif);

            fprintf(notif, "# QUICKPOOL: %s\n", "User input");
            k = 0;

            int find =FALSE;
            while (fgets(line, PATH_MAX, fp)!= NULL) {
                if (strstr(line, "# LOGIN_SHELL") != NULL)
                    continue;

                if (strstr(line, "SCRIPT_\n") != NULL) {
                    if (find == FALSE) {

                        find = TRUE;
                        continue;
                    }

                    break;
                }

                if (strstr(line, "ExitStat=$") != 0) {
                    break;
                } else {
                    if (++k > 50) {
                        fputs("\n(... %s ...)\nmore", notif);
                        break;
                    }

                    if (k == 1) {
                        remove_job_starter(line, jobStarter);
                    }

                    log_message(DEBUG, EXECUTE, "Job Command: %s", jobBase->submit->command);
                    log_message(DEBUG, EXECUTE, "Job line: %s", line);

                    fputs(line, notif);
                }
            }
            fprintf(notif, "------------------------------------------------------------\n\n");
        }

        fclose(fp);
        fp = NULL;
    }

    if (!FLOAT_EQUAL_ZERO(cpuTime)) {
        if (WIFEXITED(w_status) && !WEXITSTATUS(w_status)) {
            fprintf(notif, "Successfully completed");
        } else {
            if (WIFSIGNALED(w_status)) {
                int sig = WTERMSIG(w_status);

                if (sig > NSIG) {
                    fprintf(notif, "Exited with unknown status 0%o", job->w_status);
                } else {
                    fprintf(notif, "Exited with signal termination: %s", strsignal(sig));
                }
                if (WCOREDUMP(w_status)) {
                    fprintf(notif, ", and core dumped");
                }
            } else {
                if (jobBase->reasons & EXIT_REQUEUE) {
                    fprintf(notif, "Job was killed by user or admin, and will be re-queued with the same jobid.");
                } else {
                    fprintf(notif, "Exited with exit code %d", WEXITSTATUS(w_status));
                }
            }
        }
    } else {

        fprintf(notif, "Exited");
    }

    if (!FLOAT_EQUAL_ZERO(jobBase->lastCpuTime + cpuTime)) {
        fprintf(notif, ".\n\n%s\n\n", "Resource usage summary:");
        fprintf(notif, "    %s   :%10.2f ", "CPU time", (cpuTime + jobBase->lastCpuTime));

        fprintf(notif, "%s.\n", "sec");

        long maxMem=0, maxSwap=0;
        unsigned int i, numPid=0;
        for (i=0; i<jobBase->number_node; i++) {
            maxMem += job->max_usage[i].mem;
            maxSwap += job->max_usage[i].swap;
            numPid += job->max_usage[i].npids;
        }

        if (1024 < maxMem) {
            fprintf(notif, "    %s :%10ld MB\n", "Max Memory", (maxMem + 512) / 1024);
        } else {
            if (0 < maxMem) {
                fprintf(notif, "    %s :%10ld KB\n", "Max Memory", maxMem);
            }
        }

        if (1024 < maxSwap) {
            fprintf(notif, "    %s   :%10ld MB\n\n", "Max Swap", (maxSwap + 512) / 1024);
        } else {
            if (0 < maxSwap) {
                fprintf(notif, "    %s   :%10ld KB\n\n", "Max Swap", maxSwap);
            }
        }

        if (0 < numPid) {
            fprintf(notif, "    %s  :%10d\n", "Max Processes", numPid);
        }
    }

    if (linkStdout && (output==notif) && !errorOpeningOutputFile) {

        fprintf(notif, "\n");
        fprintf(notif, "%s\n\n", "The output (if any) is above this job summary.");
    }

    if (notif != output && strcmp(jobBase->submit->output_file, "/dev/null")) {
        if (errout == output) {
            snprintf(line, MAXMESSAGELENGTH,
                     "Read file <%s> for stdout and stderr ouput of this job.\n",
                     jobBase->submit->output_file);
        } else {
            snprintf(line, MAXMESSAGELENGTH, "Read file <%s> for stdout output of this job.\n",
                     jobBase->submit->output_file);
        }

        strncat(ps, line, MAXMESSAGELENGTH*4);
        withps = TRUE;
    }

    if ((NULL == jobBase->submit->output_file
         || 0 != strcmp(jobBase->submit->output_file,"/dev/null")) && !linkStdout) {

        snprintf(fileName, PATH_MAX, "%s.out", jobBase->submit->script);

        if ((fp = fopen(fileName, "r")) == NULL) {
            snprintf(line, MAXMESSAGELENGTH, "Unable to read output data from the stdout buffer file <%s>: your job was probably aborted prematurely.\n", fileName);
            strncat(ps, line, MAXMESSAGELENGTH*4);
            withps = TRUE;
        } else {

            if (output == notif) {
                fprintf(output, "\n");
                fprintf(output, "%s\n\n", "The output (if any) follows:");
            }

            int nItems;
            while ((nItems = fread(line, sizeof(char), PATH_MAX, fp))) {
                if (fwrite(line, sizeof(char), nItems, output) == 0) {

                    snprintf(line, MAXMESSAGELENGTH, "\nWARNING: writing output file %s failed for job %s\nError message: %s", 
                             jobid_2string(jobBase->jobid, strJobId),
                             jobBase->submit->output_file, strerror(errno));

                    if (!sendWarning) {
                        fprintf(notif, "%s", line);
                    }

                    break;
                }
            }

            fclose(fp);
            fp = NULL;
        }
    }

    if (NULL == jobBase->submit->error_file || '\0' == jobBase->submit->error_file[0]) {

        snprintf(fileName, PATH_MAX, "%s.err", jobBase->submit->script);
        if ((fp = fopen(fileName, "r")) != NULL) {
            int nItems;
            while ((nItems = fread(line, sizeof(char), PATH_MAX, fp))) {
                fwrite(line, sizeof(char), nItems, errout);
            }
            fclose(fp);
        }
    } else if (0 != strcmp(jobBase->submit->error_file, "/dev/null")) {

        snprintf(fileName, PATH_MAX, "%s.err", jobBase->submit->script);
        if (!linkStderr && (fp= fopen(fileName,"r"))== NULL) {
            snprintf(line, MAXMESSAGELENGTH, "Unable to read stderr data from stderr buffer file; your job was probably aborted prematurely.\n");
            strncat(ps, line, MAXMESSAGELENGTH*4);
            withps = TRUE;
        } else {
            if (errFileError) {
                fprintf(errout, "\n\n%s\n\n", "PS: The stderr output (if any) follows:");
            } else {
                snprintf(line, MAXMESSAGELENGTH, "Read file <%s> for stderr output of this job.\n",
                         jobBase->submit->error_file);
                strncat(ps, line, MAXMESSAGELENGTH*4);
                withps = TRUE;
            }

            int nItems;
            while (!linkStderr && (nItems = fread(line, sizeof(char), PATH_MAX, fp))) {
                fwrite(line, sizeof(char), nItems, errout);
            }

            fclose(fp);
        }
    }

    fflush(output);
    fflush(errout);
    fflush(notif);

    if (jobBase->num_exec_files && notif == output) {
        fstat(fileno(output), &stb);
    }
    ofIdx = -1;

    if (withps) {
        if (notif == output)
            fprintf(notif, "\n\nPS:\n\n%s\n", ps);
        else
            fprintf(notif, "\n%s\n", ps);
    }

    if (ofIdx >= 0) {
        if (rcp_file(&job->job_base, jobBase->exec_files+ofIdx, jobBase->submit->from_node,
                     XF_OP_EXEC2SUB, rcpMsg) < 0) {

            hasError = -1;
        } else {
            if (rcpMsg[0] != '\0') {
                log_message(DEBUG, EXECUTE, "%s: Copy output file <%s> to <%s> on submission host <%s> for job <%s>: %s",
	                    __func__, jobBase->exec_files[ofIdx].out_file,
                            jobBase->exec_files[ofIdx].in_file, jobBase->submit->from_node,
                            jobid_2string(jobBase->jobid, strJobId), rcpMsg);
            }
        }
    }

    return (hasError);
}

/**
 * @brief       To suspend the job.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, execute_suspend_command failure
 */
static int
stop_job(struct jobEntry *job, int reasons, int subreasons)
{
    int sigValue = 0;

    if (reasons & SUSPEND_REASON_USER_STOP) {
        sigValue = QUICKPOOL_SIGSTOP_USER;
    } else if (reasons & SUSPEND_REASON_QUEUE_WINDOW) {
        sigValue = QUICKPOOL_SIGSTOP_WINDOW;
    } else if (reasons & LOAD_REASONS) {
        sigValue = QUICKPOOL_SIGSTOP_LOAD;
    } else if (reasons & SUSPEND_REASON_PREEMPTED) {
        sigValue = QUICKPOOL_SIGSTOP_PREEMPT;
    } else if (reasons & SUSPEND_REASON_RESOURCE) {
        sigValue = QUICKPOOL_SIGSTOP_OTHER;
    }

    if (0 == sigValue) {
        return 0;
    }

    reasons |= SUSPEND_REASON_JMSTART;

    int status = (reasons & SUSPEND_REASON_USER_STOP) ? JOB_STATUS_USUSP : JOB_STATUS_SSUSP;

    return (execute_suspend_command(job, sigValue, status, reasons, subreasons, FALSE));
} // end function stop_job

static void
copy_jobinfo(struct jobInfo *dst, struct jobInfo *src)
{
    dst->node_execinfo = src->node_execinfo;
    dst->number_node = src->number_node;

    memcpy(&(dst->thresholds), &(src->thresholds), sizeof(struct thresholds));

    dst->exec_files = src->exec_files;
    dst->num_exec_files = src->num_exec_files;

    dst->environment = src->environment;
    dst->num_environment = src->num_environment;

    dst->eexec.len = src->eexec.len;
    dst->eexec.data = src->eexec.data;

    dst->submit = src->submit;
    dst->schedule_host = src->schedule_host;

    dst->job_file.len = src->job_file.len;
    dst->job_file.data = src->job_file.data;

    return;
}

/**
 * @brief       Set uid and gid for job pre and post script.
 *
 * @param[in]   job          #1: job
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
static int
setuid_4prepost_cmd(struct jobEntry *job)
{
    uid_t prepostUid;

    if (initgroups(job->job_base.execute_user, job->execute_gid) < 0) {
        log_message(ERR, NOCLASS, "%s: initgroups() failed for user %s uid %d gid %d %m", __func__,
                    job->job_base.execute_user, job->job_base.execute_uid, job->execute_gid);
    }

    if (setgid(job->execute_gid) < 0) {
        log_message(ERR, NOCLASS, "%s: setgid() failed for user %s uid %d gid %d %m", __func__,
                    job->job_base.execute_user, job->job_base.execute_uid, job->execute_gid);
    }

    if (NULL != job->job_base.prepost_user && '\0' != job->job_base.prepost_user[0]) {
        if (get_uid(job->job_base.prepost_user, &prepostUid) != 0) {
            prepostUid = job->job_base.execute_uid;
        }

        if (setuid(prepostUid) < 0) {
            log_message(ERR, NOCLASS, "%s: setuid() failed for uid %d gid %d %m", __func__,
                        prepostUid, job->execute_gid);
            return -1;
        }
    }

    return 0;
} // end function setuid_4prepost_cmd

/**
 * @brief       Requeue job if its exit code is one of value in REQUEUE_EXIT_VALUE.
 *
 * @param[in]   job          #1: job
 *
 * @retval      TRUE         #1: to requeue the job
 * @retval      FALSE        #2: do not requeue the job
 */
static bool_t
auto_requeue_job(struct jobEntry *job)
{
    int w_status;
    STATUS_WAIT_T status;

    JOB_WSTATUS(status) = job->w_status;
    w_status = WEXITSTATUS(status);

    struct jobInfo *jobBase = &job->job_base;
    char strJobId[32];
    log_message(DEBUG, EXECUTE, "%s: Exit status for job <%s> is <%d>, normal termination is <%d>, w_status <%d>, requeue exit code <%s>",
                __func__, jobid_2string(jobBase->jobid, strJobId), w_status, WIFEXITED(status),
                job->w_status, jobBase->requeue_exitcode?jobBase->requeue_exitcode:"NULL");

    if (!WIFEXITED(status)
        || !jobBase->requeue_exitcode || jobBase->requeue_exitcode[0] == '\0') {
        return FALSE;
    }

    char *sp;
    sp = jobBase->requeue_exitcode;

    unsigned int len = strlen(sp);
    char *value = (char *)calloc(len+1, sizeof(char));
    if (NULL == value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
        relaunch();
    }
    char *token;
    while ((token = get_string_token(&sp, " ()", value, len+1)) != NULL) {
        len = strlen(sp);

        if ((!isdigit(token[0])) || (atoi(token) != w_status)) {
            continue;
        }

        FREEUP(value);
        return TRUE;
    }
    FREEUP(value);

    return FALSE;
} // end function auto_requeue_job

#if defined(QUICKPOOL_WITH_CUDA)
/* TODO 
 * For integration with distributed pytorch training job launch.
 * Set environment variable QP_GPU_ALLOC, formatted as:
 * 2 host1 2 0,1 host2 2 0,1
 */
static int
setup_gpu_env_for_parallal_task(struct jobEntry *job)
{
    char GPU_IDX[MAXNAMELEN+1];
    char GPU_ALLOC_ENV[BUFSIZ];
    int gpuTbl[64];
    int pos = 0;
    int len = BUFSIZ;
    int i, j, k;
    int num;
    struct usedGpuIndex *gpu;

    pos += snprintf(GPU_ALLOC_ENV + pos, len - pos, "%d ", job->job_base.number_node);

    for (i = 0; i < job->job_base.number_node; i++) {
        num = 0;
        for (gpu = (struct usedGpuIndex *)job->job_base.node_execinfo[i].alloc_gpus->forw;
             gpu != (struct usedGpuIndex *)job->job_base.node_execinfo[i].alloc_gpus;
             gpu = (struct usedGpuIndex *)gpu->forw) {

            for (j = 0; j < num; j++) {
                if (gpuTbl[j] == gpu->gpu_index) {
                    break;
                }
            }
            if (j == num) {
                gpuTbl[num++] = gpu->gpu_index;
            }

            for (k = 0; k < num; k++) {
                if (k == 0) {
                    snprintf(GPU_IDX, 12, "%d", gpuTbl[k]);
                } else {
                   snprintf(GPU_IDX+strlen(GPU_IDX), MAXNAMELEN, ",%d", gpuTbl[k]);
                }
            }
        }

        pos += snprintf(GPU_ALLOC_ENV + pos, len - pos, "%s ",
                        job->job_base.node_execinfo[i].host_name);
        pos += snprintf(GPU_ALLOC_ENV + pos, len - pos, "%d ", num);
        pos += snprintf(GPU_ALLOC_ENV + pos, len - pos, "%s ", GPU_IDX);
    }

    GPU_ALLOC_ENV[strlen(GPU_ALLOC_ENV) - 1] = '\0';
    setenv("QP_GPU_ALLOC", GPU_ALLOC_ENV, 1);

    return 0;

}
#endif

static int
setup_gpu_option(struct jobEntry *job)
{
#if defined(QUICKPOOL_WITH_CUDA)
    if (0 == job->job_base.number_node) {
        return 0;
    }

    if (0 == LIST_NUM_ENTS(job->job_base.node_execinfo[0].alloc_gpus)) {
        return 0;
    } 

    if (NULL == gptrMyHost->gpuinfo_static) {
        log_message(ERR, NOCLASS, "%s: node <%s> does not have GPU installed.",
                    __func__, gptrMyHost->name);
        return -1;
    }

    if (!isnvml_initialized()) {
        log_message(ERR, NOCLASS, "%s: nvml/mxsml/dlml/cndev lib is not initialized.", __func__);
        return -1;
    }

    char strJobId[64];    

    int *index;
    index = (int *)calloc(LIST_NUM_ENTS(job->job_base.node_execinfo[0].alloc_gpus), sizeof(int));
    if (NULL == index) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                    LIST_NUM_ENTS(job->job_base.node_execinfo[0].alloc_gpus)*sizeof(int));
        relaunch();
    }

    unsigned int num = 0, m = 0;
    //the index maybe out of order because of sort_gpu_load
    unsigned int i;
    struct usedGpuIndex *gpu;
    for (gpu = (struct usedGpuIndex *)job->job_base.node_execinfo[0].alloc_gpus->forw;
         gpu != (struct usedGpuIndex *)job->job_base.node_execinfo[0].alloc_gpus;
         gpu = (struct usedGpuIndex *)gpu->forw) {

        for (m = 0; m < num; m++) {
            if (index[m] == gpu->gpu_index) {
                break;
            }
        }
        if (m == num) {
            index[num++] = gpu->gpu_index;
        }
    }

    if (isnvml_initialized()) {
        const char *GPU_ENV = "CUDA_VISIBLE_DEVICES";
        char GPU_ENV_VALUE[MAXNAMELEN+1];
        for (i = 0; i < num; i++) {
            if (i == 0) {
                snprintf(GPU_ENV_VALUE, MAXNAMELEN, "%d", index[i]);
            } else {
                snprintf(GPU_ENV_VALUE, MAXNAMELEN, "%s,%d", GPU_ENV_VALUE, index[i]);
            }
        }
    
        if (setenv(GPU_ENV, GPU_ENV_VALUE, 1)) {
            log_message(ERR, NOCLASS, "%s: job[%s] set gpu env failed",
                        __func__, jobid_2string(job->job_base.jobid, strJobId));
            FREEUP(index);
            return -1;
        }

        /* TODO
         * Set GPU environment variables for distributed pytorch job integration.
         */
        setup_gpu_env_for_parallal_task(job);
    }

    FREEUP(index);
#endif

    return 0;
}

/* bind_to_num_mem()
 */
static int 
bind_to_numa_mem(struct jobEntry *job)
{
#if defined (HAVE_HWLOC_H)
    const struct hwloc_topology_support *numaSupport;
    numaSupport = hwloc_topology_get_support(topology);

    if (!numaSupport->membind->bind_membind || !numaSupport->membind->set_thisproc_membind){
        log_message(ERR, NOCLASS, "%s: memory binding failed,because OS don't support", __func__);  
        return -1;
    }

    struct nodeExecution *myExec = &(job->job_base.node_execinfo[0]);
    if (0 == LIST_NUM_ENTS(myExec->alloc_cores)) {
        log_message(ERR, NOCLASS, "%s: memory binding failed,because numaInfo is NULL", __func__); 
        return -1;
    }
    
    int type = job->job_base.membind_type;
    
    hwloc_bitmap_t nodeSet;
    nodeSet = hwloc_bitmap_alloc();
    if (NULL == nodeSet) {
        log_message(ERR, NOCLASS, "%s: memory binding failed,because nodeSet alloc failed", __func__);
        return -1;
    }

    struct usedCoreInfoDown *core;
    for (core = (struct usedCoreInfoDown *)myExec->alloc_cores->forw;
         core != (struct usedCoreInfoDown *)myExec->alloc_cores;
         core = (struct usedCoreInfoDown *)core->forw) {

        hwloc_bitmap_set(nodeSet, core->index);
    }
    
    int ret = 0;
    /*maybe the mem binding policy can chose HWLOC_MEMBIND_INTERLEAVE replace HWLOC_MEMBIND_BIND,
    HWLOC_MEMBIND_INTERLEAVE can balance the memory references */
    if (job->job_base.membind_type == MEMBIND_ONLY) {
        if (!numaSupport->membind)
            ret = hwloc_set_membind(topology, (hwloc_const_bitmap_t)nodeSet, HWLOC_MEMBIND_BIND,
                                    HWLOC_MEMBIND_PROCESS | HWLOC_MEMBIND_BYNODESET | HWLOC_MEMBIND_STRICT);
    } else if (job->job_base.membind_type == MEMBIND_PREFER) {
        ret = hwloc_set_membind(topology, (hwloc_const_bitmap_t)nodeSet, HWLOC_MEMBIND_BIND,
                                HWLOC_MEMBIND_PROCESS | HWLOC_MEMBIND_BYNODESET);
    } else {

        log_message(ERR, NOCLASS, "%s: memory binding policy is unknown type[%d]", __func__, type);  
        hwloc_bitmap_free(nodeSet);
        return -1;
    }

    hwloc_bitmap_free(nodeSet);
    if (ret) {
        log_message(ERR, NOCLASS, "%s: memory binding failed by call hwloc_set_membind", __func__); 
        return -1;
    }
#endif     
    return 0;
} 

/* bind_to_numa_core()
 */
static int
bind_to_numa_core(struct jobEntry *job)
{
#if defined (HAVE_HWLOC_H)
    const struct hwloc_topology_support *numaSupport;
    numaSupport = hwloc_topology_get_support(topology);

    if (!numaSupport->cpubind->set_thisproc_cpubind){
        log_message(ERR, NOCLASS, "%s: cpubind binding failed,because OS don't support", __func__);  
        return -1;
    }

    struct nodeExecution *myExec = &(job->job_base.node_execinfo[0]);
    if (0 == LIST_NUM_ENTS(myExec->alloc_cores)) {
        log_message(ERR, NOCLASS, "%s: memory binding failed,because numaInfo is NULL", __func__); 
        return -1;
    }

    hwloc_bitmap_t cpuSet = NULL;
    cpuSet = hwloc_bitmap_alloc();
    if (!cpuSet) {
        log_message(ERR, NOCLASS, "%s: memory binding failed,because nodeSet alloc failed", __func__);
        return -1;
    }

    struct usedCoreInfoDown *core;
    for (core = (struct usedCoreInfoDown *)myExec->alloc_cores->forw;
         core != (struct usedCoreInfoDown *)myExec->alloc_cores;
         core = (struct usedCoreInfoDown *)core->forw) {

        hwloc_bitmap_set(cpuSet, core->index);
    }

    /* bind process pid on cores specified by cpuset */
    /*flag :
        HWLOC_CPUBIND_PROCESS 1  bind all  proccess or thread under the pid 
        HWLOC_CPUBIND_THREAD  2  bind the proccess or thread only
        HWLOC_CPUBIND_STRICT  4  strict binding,if the binding cpu busy, os will not 
                                    despatch the proccess to other free cpu
        HWLOC_CPUBIND_NOMEMBIND 8  normally, the function will binding the memory when 
                                    binding the cpu, this flag will not binding memory
        */
    if (hwloc_set_cpubind(topology, cpuSet, HWLOC_CPUBIND_PROCESS)) {
        log_message(ERR, NOCLASS, "%s: cpu binding failed by call hwloc_set_cpubind", __func__);
        hwloc_bitmap_free(cpuSet);
        return -1; 
    }
    hwloc_bitmap_free(cpuSet);
#endif    
    return 0;
}

static int
setup_cpu_bind(struct jobEntry *job)
{
    if (0 == LIST_NUM_ENTS(job->job_base.node_execinfo[0].alloc_cores)) {
        return 0;
    }   

    char strJobId[64];    
    char myhost[MAXHOSTNAMELEN];
    if (gethostname(myhost, MAXHOSTNAMELEN) <0) {
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "gethostname", myhost);
        return -1;
    }
    
    if (bind_to_numa_core(job)) {
        log_message(ERR, NOCLASS, "%s: bind numa core failed jobId[%s]",
                    __func__, jobid_2string(job->job_base.jobid, strJobId));
        return -1;
    }
    
    if (!job->job_base.membind_type) {
        return 0;
    }   

    if (bind_to_numa_mem(job)) {
        log_message(ERR, NOCLASS, "%s: bind numa core failed jobId[%s]",
                    __func__, jobid_2string(job->job_base.jobid, strJobId));
        return -1;
    }
   
    return 0;
}
